mcpField
A
agentic-microservice-deployer

agentic-microservice-deployer

github.com/kevinrivm云与 DevOps336 次安装29 周前更新
代码仓库

关于 agentic-microservice-deployer

此技能指导 Agent 将本地自动化代码构建为部署在 Coolify 的内部微服务,并通过 Docker 内部网络供 n8n 访问。它涵盖 GitHub 仓库、部署、测试和端点文档流程。

主要功能

  1. 01指导收集真实测试数据
  2. 02创建并配置 GitHub 仓库
  3. 03在 Coolify 部署内部服务
  4. 04监控部署并运行自测
  5. 05生成 n8n 端点文档

适用场景

  1. 01将本地自动化封装为内部 API
  2. 02为 n8n 工作流部署服务
  3. 03验证端点并生成调用说明

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/kevinrivm/agentic-microservice-deployer --skill agentic-microservice-deployer

安装后在 Claude Code 中输入/agentic-microservice-deployer确认已加载。

包含的文件

除 SKILL.md 外,安装时一起下载的参考资料和脚本。

README.md
.gitattributes
execution/
coolify_manager.py
github_manager.py

详细介绍

Agentic Microservice Deployer

将本地自动化代码转换为安全的内部微服务,部署在 Coolify 上, 并仅通过 Docker 内部网络供 n8n 访问。没有公开 URL。无需繁琐操作。


⚙️ 初始设置 — 开始前必读

此 skill 包含两个执行脚本,必须存在于项目的 execution/ 目录中。 如果不存在,请从 skill 文件夹复制创建:

import shutil, os

# Ruta donde está instalado el skill (ajustar si es diferente)
SKILL_DIR = ".agent/skills/agentic-microservice-deployer/execution"

os.makedirs("execution", exist_ok=True)
for script in ["coolify_manager.py", "github_manager.py"]:
    src = os.path.join(SKILL_DIR, script)
    dst = os.path.join("execution", script)
    if not os.path.exists(dst):
        if os.path.exists(src):
            shutil.copy(src, dst)
            print(f"Copiado: {script}")
        else:
            print(f"ADVERTENCIA: {src} no encontrado. Descarga manual necesaria.")
    else:
        print(f"Ya existe: execution/{script}")

如果尚未安装依赖项,也请安装:

pip install requests python-dotenv

如果 skill 文件夹中没有这些脚本,可以从仓库获取:

https://github.com/kevinrivm/agentic-microservice-deployer/tree/main/execution

手动与自动操作一览

步骤执行者频率
在 Coolify UI 中创建 GitHub App👤 用户每个 VPS 1 次
其他所有操作🤖 Agent每个项目

唯一的手动步骤需要浏览器,因为 GitHub 需要用户点击“Install” (这是 GitHub 有意设置的安全 OAuth 握手流程)。没有可替代它的 API。


步骤 0:收集测试数据 — 构建前必须完成

为什么要先做这一步? Agent 需要真实的测试数据,以便:

  1. 了解准确的输入格式(避免因猜测而生成错误代码)
  2. 在交付给用户之前对服务进行自动测试
  3. 从一开始就验证边界情况

至少取得一个测试示例后,再开始编写代码。

继续之前,务必先询问用户:


构建微服务之前,我需要真实的测试数据。 这将使我能够:

  • 根据你的数据实际格式构建代码
  • 在交付服务之前对其进行自动测试
  • 在你发现错误之前找出问题

请提供:

  1. 服务将接收的每种输入类型的一个真实示例 (例如:一张示例 OCR 收据、一封完整邮件、一张图片或一份 JSON 示例)
  2. 该输入对应的预期响应 (例如:“我希望它从这张收据中提取:日期、总额、供应商”)
  3. 一个边界情况或困难案例(如果有) (例如:“收据有时会模糊不清或是英文的”)

如果你现在没有可用的真实数据,我可以根据你的描述生成合成数据,但真实数据能带来更好的结果。


取得测试数据后:

  • 将其保存在 .tmp/test_data/,以便开发期间参考
  • 以其为基础构建 Pydantic 模型和服务逻辑
  • 在交付前的**步骤 5(自动测试)**中使用同一批数据

前置条件 — 开始前检查

环境变量(在 .env 中):
GITHUB_TOKEN=         # PAT con scope 'repo' — crear repos privados + vincular GitHub Apps
COOLIFY_URL=          # ej: https://panel-coolify.tudominio.com
COOLIFY_TOKEN=        # API token de Coolify (Settings > API Keys)
COOLIFY_PROJECT_UUID= # UUID del proyecto Coolify donde desplegar

如果缺少任何变量,请先向用户索取,再继续。

Coolify 中的 GitHub App(每个 VPS 1 次):

如果用户从未在 Coolify 中配置过此项,请向其说明:

为什么需要手动操作? GitHub 要求用户在浏览器中点击“Install” (OAuth 握手)。每个 VPS/客户端只需操作一次。之后,Agent 就能管理所有未来仓库的访问权限,无需用户介入。

用户操作说明(5 分钟):


  1. Coolify 面板 → 侧边菜单 “Sources”
  2. 点击 “Add” → “GitHub App”
  3. Name:输入描述性名称(例如:coolify-tudominio)
  4. Organization:留空
  5. System Wide:不要勾选
  6. Webhook Endpoint:选择 Coolify 的 HTTPS 域名
  7. 点击 “Register Now” → GitHub 将打开安装页面
  8. 在 GitHub 中:选择 “All repositories” ← 关键操作,确保 Agent 能管理后续的访问权限
  9. 点击 “Install” → 自动返回 Coolify

✅ 完成后,Agent 就能自主管理所有未来的仓库。


确认完成后,继续执行步骤 1。


步骤 1:准备仓库结构

生产仓库只包含容器所需的内容。

所需的最小结构:

nombre-del-servicio/
├── main.py               # FastAPI con X-API-Key y GET /health
├── execution/            # scripts usados en runtime
├── directives/
│   └── README.md         # qué hace el servicio, endpoints, variables
├── Dockerfile
├── requirements.txt
├── .env.example
└── .gitignore

必需的 .gitignore:

.env
.env.*
!.env.example
*.json
!package.json
__pycache__/
*.pyc
.tmp/
credentials.json
token.json

标准 Dockerfile:

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

包含身份验证的最小 main.py:

from fastapi import FastAPI, HTTPException, Security
from fastapi.security import APIKeyHeader
import os

app = FastAPI()
API_KEY = os.getenv("SERVICE_API_KEY", "")
api_key_header = APIKeyHeader(name="X-API-Key")

async def verify_key(key: str = Security(api_key_header)):
    if key != API_KEY:
        raise HTTPException(status_code=403, detail="Forbidden")
    return key

@app.get("/health")
def health():
    return {"status": "ok"}

# Endpoints del negocio — siempre con dependencies=[Depends(verify_key)]

步骤 2:在 GitHub 上创建仓库并关联 GitHub App

使用 execution/github_manager.py:

import sys, os, requests
sys.path.insert(0, 'execution')
from github_manager import create_private_repo, grant_github_app_access, initialize_and_push
from dotenv import load_dotenv
load_dotenv()

REPO_NAME   = "nombre-del-servicio"
DESCRIPTION = "Microservicio: descripción breve"

# 2a. Crear repo privado
result = create_private_repo(REPO_NAME, DESCRIPTION)
if not result:
    raise Exception("Fallo la creacion del repo")
owner, repo_name, repo_id = result

# 2b. Vincular al GitHub App de Coolify (automatiza el paso de "dar acceso al repo")
COOLIFY_URL   = os.getenv("COOLIFY_URL")
COOLIFY_TOKEN = os.getenv("COOLIFY_TOKEN")
coolify_h = {"Authorization": f"Bearer {COOLIFY_TOKEN}", "Accept": "application/json"}
github_apps = requests.get(f"{COOLIFY_URL}/api/v1/github-apps", headers=coolify_h).json()
private_apps = [a for a in github_apps if not a.get("is_public")]
installation_id = private_apps[0]["installation_id"]
grant_github_app_access(repo_id, installation_id)

# 2c. Subir el código
initialize_and_push(repo_name, owner)

步骤 3:在 Coolify 中创建并配置应用

⚠️ 以下 3 个子步骤均为必需步骤。若跳过 3b,服务将公开暴露, 且 n8n 无法通过网络名称访问该服务。

3a. 创建应用
import sys
sys.path.insert(0, 'execution')
from coolify_manager import CoolifyManager

manager = CoolifyManager()
ALIAS = repo_name   # nombre DNS interno — n8n usará: http://ALIAS:8000

# Endpoint correcto: /api/v1/applications/private-github-app
# (NO /api/v1/applications — ese da 404 para repos privados)
app = manager.create_application(repo_name, f"{owner}/{repo_name}")
APP_UUID = app["uuid"]
print(f"App creada: {APP_UUID}")
3b. 配置内部网络——至关重要,切勿跳过

Coolify 会自动分配公共 FQDN。此步骤会将其配置为内部服务:

manager.configure_application(APP_UUID, ALIAS)
# Confirma: alias de red configurado, FQDN eliminado, healthcheck activo

重要: 如果 configure_application 无法通过 API 删除 FQDN (某些版本的 Coolify 不允许这样做),请告知用户手动删除: 面板 → 应用 → Settings → FQDN → 删除 → Save。 即使存在公共 FQDN,该服务始终可通过内部网络访问, 但出于安全考虑,仍应将其删除。

3c. 用户是否需要使用 API key 进行身份验证?

配置变量之前,请先询问用户:

该服务是内部服务(只能通过 Docker 网络从 n8n 访问)。 网络已提供保护——API key 是可选的额外安全层。

你想添加 X-API-Key 身份验证吗?

  • 是 → 可更好地防范未来的错误配置或横向入侵
  • 否 → 更简单,n8n 无需额外请求头即可调用

如果用户回答“是”——使用 API key:

import requests, os, secrets
from dotenv import load_dotenv
load_dotenv()

SERVICE_API_KEY = secrets.token_hex(32)  # NUNCA usar strings hardcodeados
print(f"SERVICE_API_KEY: {SERVICE_API_KEY}")
print("(Guarda este valor — lo usarás en n8n como header X-API-Key)")

COOLIFY_URL = os.getenv("COOLIFY_URL")
headers = {"Authorization": f"Bearer {os.getenv('COOLIFY_TOKEN')}", "Content-Type": "application/json"}

env_vars = [
    {"key": "SERVICE_API_KEY", "value": SERVICE_API_KEY},
    # Otras vars del servicio:
    # {"key": "OPENAI_API_KEY", "value": "sk-..."},
]
for var in env_vars:
    requests.post(f"{COOLIFY_URL}/api/v1/applications/{APP_UUID}/envs",
                  headers=headers, json={**var, "is_preview": False})
print("Variables configuradas")

确保 main.py 中启用了身份验证中间件(参见步骤 1)。

如果用户回答“否”——不使用 API key:

# Solo configurar las vars de negocio (sin SERVICE_API_KEY)
env_vars = [
    # {"key": "OPENAI_API_KEY", "value": "sk-..."},
]
for var in env_vars:
    requests.post(f"{COOLIFY_URL}/api/v1/applications/{APP_UUID}/envs",
                  headers=headers, json={**var, "is_preview": False})

在 main.py 中,简化端点,删除 Security(api_key_header):

@app.post("/mi-endpoint")
def mi_endpoint(data: dict):  # sin dependencies=[Depends(verify_key)]
    ...
3d. 部署
result = manager.deploy_application(APP_UUID)
deployment_uuid = result["deployments"][0]["deployment_uuid"]
print(f"Deploy iniciado: {deployment_uuid}")

步骤 4:监控部署

import time, json, requests, os
from dotenv import load_dotenv
load_dotenv()

headers = {"Authorization": f"Bearer {os.getenv('COOLIFY_TOKEN')}", "Accept": "application/json"}
time.sleep(90)

r = requests.get(f"{os.getenv('COOLIFY_URL')}/api/v1/deployments/{deployment_uuid}", headers=headers)
data = r.json()
status = data.get("status")
logs = json.loads(data.get("logs", "[]"))

print(f"Status: {status}")
for log in logs:
    if not log.get("hidden"):
        output = log.get("output", "")[:200]
        if output.strip():
            print(f"  [{log.get('type','').upper()}] {output}")

成功标志:

  • status: finished
  • 日志:Healthcheck status: "healthy"
  • 日志:Rolling update completed

步骤 5:交付用户前进行自动测试

为什么先使用公共 FQDN 测试? Agent 在用户的机器(主机)上运行,而不是在 Docker 内部。 内部 URL http://alias:8000 只能从 coolify 网络中的容器访问。 为了让 Agent 能够实际发送 HTTP 测试请求,需要使用临时公共 URL。

5a. 向用户索取测试数据

测试之前,请询问用户:

“为了在交付前确认微服务正常运行,我需要每个端点的真实测试数据。例如:

  • 你希望我处理哪个测试邮箱?
  • 你希望发送什么示例文本/payload?

如果你现在没有真实数据,我可以根据架构生成合成数据。”

5b. 获取刚完成部署的公共 URL
import requests, os, time
from dotenv import load_dotenv
load_dotenv()

headers_api = {"Authorization": f"Bearer {os.getenv('COOLIFY_TOKEN')}", "Accept": "application/json"}
app = requests.get(f"{os.getenv('COOLIFY_URL')}/api/v1/applications/{APP_UUID}", headers=headers_api).json()

PUBLIC_URL = app.get("fqdn", "").rstrip("/")
print(f"URL publica para tests: {PUBLIC_URL}")
# ej: http://uuid.187.77.17.72.sslip.io

注意: FQDN 仅在测试期间保持启用。Coolify 会在创建应用时自动分配该域名。所有测试通过后,我们会在步骤 5d 中将其删除。

5c. 执行测试
import json

AUTH_HEADER = {"X-API-Key": SERVICE_API_KEY} if SERVICE_API_KEY else {}
test_results = []

# Test 1: Healthcheck (siempre)
r = requests.get(f"{PUBLIC_URL}/health", headers=AUTH_HEADER, timeout=10)
passed = r.status_code == 200 and r.json().get("status") == "ok"
test_results.append({"test": "GET /health", "status": r.status_code, "passed": passed})
print(f"{'OK' if passed else 'FAIL'} GET /health → {r.status_code}")

# Test 2+: Endpoints del negocio (usar data de prueba del usuario o sintética)
# Ejemplo — adaptar según los endpoints reales de main.py:
test_payload = {
    # poner aqui los datos de prueba que dio el usuario
}
r2 = requests.post(
    f"{PUBLIC_URL}/tu-endpoint",
    headers={**AUTH_HEADER, "Content-Type": "application/json"},
    json=test_payload,
    timeout=30
)
passed2 = r2.status_code == 200
test_results.append({"test": "POST /tu-endpoint", "status": r2.status_code,
                     "passed": passed2, "response": r2.text[:200]})
print(f"{'OK' if passed2 else 'FAIL'} POST /tu-endpoint → {r2.status_code}")
print(f"  Response: {r2.text[:200]}")

# Resumen
all_passed = all(t["passed"] for t in test_results)
print(f"\n{'TODOS LOS TESTS PASARON' if all_passed else 'HAY TESTS FALLIDOS'}")
for t in test_results:
    print(f"  {'OK' if t['passed'] else 'FAIL'} {t['test']}")

如果有任何测试失败:

  • 查看部署日志,了解错误原因
  • 在 main.py 中修正代码
  • 执行 git push + manager.deploy_application(APP_UUID),等待完成后重新测试
  • 重复以上步骤,直到所有测试通过
5d. 测试通过 → 迁移到内部网络

所有测试通过后,删除 FQDN 并将其配置为内部服务:

base = f"{os.getenv('COOLIFY_URL')}/api/v1/applications/{APP_UUID}"
headers_coolify = {"Authorization": f"Bearer {os.getenv('COOLIFY_TOKEN')}",
                   "Content-Type": "application/json"}

# PATCHes separados (la API de Coolify rechaza campos mezclados)
requests.patch(base, headers=headers_coolify, json={"custom_network_aliases": ALIAS})
requests.patch(base, headers=headers_coolify, json={"domains": ""})  # elimina FQDN publico
requests.patch(base, headers=headers_coolify, json={"dockerfile_location": "/Dockerfile"})

# Redeploy para aplicar cambio de red
manager.deploy_application(APP_UUID)
time.sleep(60)

# Verificar
app_final = requests.get(base, headers=headers_coolify).json()
assert not app_final.get("fqdn"), "FQDN no eliminado"
assert app_final.get("custom_network_aliases") == ALIAS, "Alias no configurado"
print(f"Servicio migrado a red interna. URL para n8n: http://{ALIAS}:8000")

步骤 6:生成端点文档(API_DOCS.md)

Agent 会读取 main.py,并生成包含 curl 命令和 n8n 就绪配置的文件。 用户可将内容直接复制到 HTTP Request 节点中,无需修改代码。

# Leer main.py para extraer endpoints reales
with open("main.py", "r", encoding="utf-8") as f:
    source = f.read()

# El agente lee el codigo y extrae: metodo HTTP, ruta, descripcion (docstring),
# campos del body (modelos Pydantic), respuesta esperada.
# Genera una entrada por endpoint. NO usar valores genéricos.

PORT = 8000
auth_line = f"X-API-Key: {SERVICE_API_KEY}" if SERVICE_API_KEY else "(sin autenticacion)"

doc = f"""# API Documentation — {ALIAS}

**URL base (desde n8n):** `http://{ALIAS}:{PORT}`
**Autenticacion:** {auth_line}

---

## Endpoints

### GET /health
Verifica que el servicio está activo.

curl:
  curl http://{ALIAS}:{PORT}/health -H 'X-API-Key: {SERVICE_API_KEY}'

n8n HTTP Request:
  Method: GET
  URL: http://{ALIAS}:{PORT}/health
  Header X-API-Key: {SERVICE_API_KEY}

Respuesta: {{"status": "ok"}}

---

[El agente agrega aqui una seccion por cada @app.post/@app.get encontrado en main.py,
con el payload real basado en los modelos Pydantic definidos]

---

## Conectar a n8n

1. Crea un workflow en n8n
2. Agrega: [trigger real] → HTTP Request → [logica adicional]
3. Copia la config del endpoint desde este doc al nodo HTTP Request
4. El microservicio ya está disponible en la red interna de Coolify
"""

with open("API_DOCS.md", "w", encoding="utf-8") as f:
    f.write(doc)
print("API_DOCS.md generado — compartir con el usuario.")

安全:标准规则

规则描述
🔒 始终使用私有仓库绝不将客户代码放在公开仓库中
🧪 测试期间使用临时 FQDN所有测试通过后将其删除
🔑 SERVICE_API_KEY 可选用户需要时使用 secrets.token_hex(32);否则省略
📁 .env 仅用于本地Coolify 通过 API 注入变量
🏷️ 使用描述性别名客户-服务,例如:acme-yt-optimizer

故障排除

“没有已连接的私有 GitHub Apps” → 用户尚未创建 GitHub App。请参阅“前置条件”一节。

grant_github_app_access 返回 404 → installation_id 没有“所有仓库”权限。用户需要前往 GitHub → Settings → Applications → [你的 GitHub App] → Repository access → All repositories。

create_application 返回 422 / validation error → git_repository 必须严格为 owner/repo(不带 git@、.git 或 https://)。

custom_network_aliases 未生效 / n8n 无法访问服务 → 如果在单个 PATCH 中发送多个字段,Coolify API 会拒绝请求。 → 解决方法是每个字段单独发送一个 PATCH:

base = f"{COOLIFY_URL}/api/v1/applications/{APP_UUID}"
requests.patch(base, headers=headers, json={"custom_network_aliases": ALIAS})
requests.patch(base, headers=headers, json={"domains": ""})
requests.patch(base, headers=headers, json={"dockerfile_location": "/Dockerfile"})

公开 FQDN 未消失 / 使用 fqdn 时出现 "This field is not allowed" → 删除公开 URL 的正确字段是 domains(不是 fqdn)。 → 使用:requests.patch(base, headers=headers, json={"domains": ""}) → 然后重新部署,让 Traefik 更新配置。

SERVICE_API_KEY 被硬编码(例如:algo-secret-key-2024) → 无法通过单独的 PATCH 更新重复变量(会返回 404)。 → 解决方法:删除旧环境变量,再 POST 新变量:

envs = requests.get(f"{COOLIFY_URL}/api/v1/applications/{APP_UUID}/envs", headers=headers).json()
for e in [x for x in envs if x["key"] == "SERVICE_API_KEY"]:
    requests.delete(f"{COOLIFY_URL}/api/v1/applications/{APP_UUID}/envs/{e['uuid']}", headers=headers)
import secrets
new_key = secrets.token_hex(32)
requests.post(f"{COOLIFY_URL}/api/v1/applications/{APP_UUID}/envs",
              headers=headers, json={"key": "SERVICE_API_KEY", "value": new_key})
print(f"Nueva key: {new_key}")

第 5 步期间测试失败——服务返回错误 500 → 查看日志:GET /api/v1/deployments/{deployment_uuid} → logs 字段 → 修复 main.py,推送代码,重新部署,等待后再次测试。 → 在所有测试通过前,不要迁移到内部网络。

健康检查失败 / running:unknown → main.py 必须包含返回 200 的 GET /health。 → 必须配置 dockerfile_location(/Dockerfile)。 → 使用 configure_application(),它会自动设置这些值。

n8n 无法访问服务(内部网络) → n8n 必须安装在同一个 Coolify 上(位于同一个 Docker 网络 coolify 中)。 → 使用 http://alias:8000——绝不要使用 localhost 或服务器 IP。 → 如果 n8n 位于另一台服务器上,则需要 FQDN 和持久的强身份验证。

评论

支持 Markdown