Files
MP-Manager/mcp_server/manifest.py
T
2026-05-30 14:31:19 -06:00

98 lines
4.4 KiB
Python

"""Genera `generated/agent/tools_manifest.json`.
Fuente única de verdad navegable para LLM/humanos: lista de tools MCP,
scripts disponibles, endpoints FastAPI y su estado de cumplimiento.
Se ejecuta automáticamente al arrancar el server. También puede invocarse
manual: python -m mcp_server.manifest
"""
from __future__ import annotations
import json
import os
import sys
from datetime import datetime, timezone
_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _ROOT not in sys.path:
sys.path.insert(0, _ROOT)
import paths # noqa: E402
# Catálogo declarativo de tools MCP (mantener sincronizado con server.py).
TOOLS = [
{"name": "list_accounts", "category": "accounts", "mutates": False, "desc": "Lista cuentas Marca+sucursales."},
{"name": "get_account", "category": "accounts", "mutates": False, "desc": "Detalle de una cuenta."},
{"name": "get_global_metrics", "category": "metrics", "mutates": False, "desc": "Métricas globales."},
{"name": "get_account_metrics", "category": "metrics", "mutates": False, "desc": "Métricas por sucursal."},
{"name": "search_contacts", "category": "contacts", "mutates": False, "desc": "Busca contactos en cache local."},
{"name": "get_contact", "category": "contacts", "mutates": False, "desc": "Detalle de contacto."},
{"name": "get_opportunities", "category": "opps", "mutates": False, "desc": "Oportunidades por location."},
{"name": "get_pipelines", "category": "opps", "mutates": False, "desc": "Pipelines/etapas por location."},
{"name": "get_workflows", "category": "workflows", "mutates": False, "desc": "Workflows por location."},
{"name": "sync_missing_contacts", "category": "sync", "mutates": True, "desc": "Sucursal→Marca contactos faltantes (dry-run default)."},
{"name": "sync_missing_opps", "category": "sync", "mutates": True, "desc": "Sucursal→Marca opps faltantes (dry-run default)."},
{"name": "sync_logs", "category": "ops", "mutates": False, "desc": "Logs de sincronización."},
{"name": "error_logs", "category": "ops", "mutates": False, "desc": "Errores recientes."},
{"name": "agent_audit_report", "category": "ops", "mutates": False, "desc": "Reporte de salud agentica."},
{"name": "script_catalog", "category": "ops", "mutates": False, "desc": "Manifest navegable (este archivo)."},
{"name": "run_script", "category": "advanced", "mutates": True, "desc": "Ejecuta script arbitrario de scripts/."},
]
def _load_audit_report() -> dict | None:
if not os.path.exists(paths.AGENT_AUDIT_REPORT):
return None
try:
with open(paths.AGENT_AUDIT_REPORT, "r", encoding="utf-8") as f:
return json.load(f)
except (OSError, json.JSONDecodeError):
return None
def build_manifest() -> dict:
audit = _load_audit_report()
scripts: list[dict] = []
endpoints: list[dict] = []
if audit:
for s in audit.get("scripts", []):
scripts.append({
"name": s["name"],
"category": s["category"],
"is_mutator": s["is_mutator"],
"registered_in_metadata": s["registered_in_metadata"],
"has_json": s["has_json_flag"],
"has_apply": s["has_apply_flag"],
"has_run_id": s["has_run_id_flag"],
"issues": s["issues"],
"suggestion": s["suggestion"],
"docstring": s["docstring"],
})
endpoints = audit.get("endpoints", [])
return {
"generated_at": datetime.now(timezone.utc).isoformat(),
"tools": TOOLS,
"scripts": scripts,
"endpoints": endpoints,
"notes": {
"apply_confirm_token": "I-HAVE-USER-CONFIRMATION",
"dry_run_default": True,
"audit_source": "generated/agent/audit_report.json",
"refresh_audit": "python scripts/audit_agent_readiness.py",
},
}
def write_manifest() -> str:
os.makedirs(paths.AGENT_DIR, exist_ok=True)
manifest = build_manifest()
with open(paths.AGENT_MANIFEST_PATH, "w", encoding="utf-8") as f:
json.dump(manifest, f, indent=2, ensure_ascii=False)
return paths.AGENT_MANIFEST_PATH
if __name__ == "__main__":
path = write_manifest()
print(f"manifest written: {path}")