cristiano.hoshikawa/agent_platform_oci
1
2
Fork 0
mirror of https://github.com/hoshikawa2/agent_platform_oci.git synced 2026-07-09 14:04:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

Ajustes na documentação e remanejamento dos folders

Browse Source
This commit is contained in:
Cristiano Hoshikawa
2026-06-21 10:25:02 -03:00
parent 8659eb552a
commit fe194a9f2a
10 changed files with 1430 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

222
apps/mcp_gateway/app/main.py
View File

@@ -45,14 +45,25 @@ class ToolResult(BaseModel):
metadata: dict[str, Any] = Field(default_factory=dict)
class DiscoveryResult(BaseModel):
ok: bool
servers_scanned: int = 0
tools_discovered: int = 0
errors: list[dict[str, Any]] = Field(default_factory=list)
tools: list[str] = Field(default_factory=list)
def load_config() -> dict[str, Any]:
path = Path(os.getenv("MCP_GATEWAY_CONFIG_PATH", "config/mcp_gateway.yaml"))
return yaml.safe_load(path.read_text(encoding="utf-8")) or {}
config = load_config()
static_tools: dict[str, dict[str, Any]] = dict(config.get("tools") or {})
discovered_tools: dict[str, dict[str, Any]] = {}
discovery_state: dict[str, Any] = {"last_sync": None, "errors": [], "tools": []}
cache: dict[str, tuple[float, Any]] = {}
app = FastAPI(title="Agent Platform OCI - MCP Gateway", version="1.0.0")
app = FastAPI(title="Agent Platform OCI - MCP Gateway", version="1.1.0")
def audit(name: str, payload: dict[str, Any]) -> None:
@@ -70,6 +81,13 @@ def auth_check(authorization: str | None) -> None:
raise HTTPException(status_code=403, detail="Invalid MCP Gateway token")
def all_tools() -> dict[str, dict[str, Any]]:
merged = dict(discovered_tools)
# Static config wins over discovery so operators can override metadata safely.
merged.update(static_tools)
return merged
def map_arguments(tool_name: str, args: dict[str, Any], bc: dict[str, Any]) -> dict[str, Any]:
result = dict(args or {})
for source, target in ((config.get("parameter_mapping") or {}).get(tool_name) or {}).items():
@@ -100,16 +118,22 @@ async def post_with_retry(url: str, payload: dict[str, Any], timeout: int, retry
raise RuntimeError(str(last_exc))
def build_server_payload(tool_name: str, args: dict[str, Any], server: dict[str, Any], tool: dict[str, Any]) -> dict[str, Any]:
"""Builds the payload expected by the downstream MCP server.
async def get_json(url: str, timeout: int) -> Any:
async with httpx.AsyncClient(timeout=timeout) as client:
resp = await client.get(url)
resp.raise_for_status()
return resp.json()
The example servers under mcp/servers expose the framework legacy contract:
POST /mcp/tools/call {tool_name, arguments}. Some mocks expose direct
tool endpoints that accept only the argument object. The gateway supports
both shapes through the server/tool config.
def build_server_payload(tool_name: str, args: dict[str, Any], server: dict[str, Any], tool: dict[str, Any]) -> dict[str, Any]:
"""Build the payload expected by the downstream MCP server.
Supported protocols:
- legacy_http/framework_http: POST /mcp/tools/call with {tool_name, arguments}
- direct_http: POST to the configured endpoint with only the argument object
"""
protocol = str(tool.get("protocol") or server.get("protocol") or "legacy_http")
if protocol in {"legacy_http", "framework_http"}:
if protocol in {"legacy_http", "framework_http", "fastmcp_http"}:
return {"tool_name": tool_name, "arguments": args or {}}
return args or {}
@@ -121,36 +145,202 @@ def normalize_server_response(data: Any) -> tuple[bool, Any, str | None, dict[st
return True, data, None, {}
def _tool_name(raw: dict[str, Any]) -> str | None:
return raw.get("name") or raw.get("tool_name") or raw.get("id")
def _tool_schema(raw: dict[str, Any]) -> dict[str, Any]:
schema = raw.get("input_schema") or raw.get("inputSchema") or raw.get("schema") or {}
if isinstance(schema, dict):
return schema
return {}
def _extract_tools_from_catalog(catalog: Any) -> list[dict[str, Any]]:
"""Normalize common MCP/FastMCP/custom catalog shapes.
Accepted examples:
- {"tools": [{"name": "x", "description": "...", "input_schema": {...}}]}
- [{"name": "x", "inputSchema": {...}}]
- {"server_id": "s", "capabilities": {"tools": [...]}}
- {"data": {"tools": [...]}}
"""
if isinstance(catalog, list):
return [x for x in catalog if isinstance(x, dict)]
if not isinstance(catalog, dict):
return []
if isinstance(catalog.get("tools"), list):
return [x for x in catalog["tools"] if isinstance(x, dict)]
data = catalog.get("data")
if isinstance(data, dict) and isinstance(data.get("tools"), list):
return [x for x in data["tools"] if isinstance(x, dict)]
caps = catalog.get("capabilities")
if isinstance(caps, dict) and isinstance(caps.get("tools"), list):
return [x for x in caps["tools"] if isinstance(x, dict)]
return []
def _catalog_urls(server_id: str, server: dict[str, Any]) -> list[str]:
if server.get("manifest_url"):
return [str(server["manifest_url"])]
base = str(server.get("url", "")).rstrip("/")
if server.get("catalog_endpoint"):
return [f"{base}{server['catalog_endpoint']}"]
discovery = config.get("discovery") or {}
endpoints = server.get("discovery_endpoints") or discovery.get("default_catalog_endpoints") or [
"/.well-known/mcp-server.json",
"/manifest",
"/mcp/tools",
"/tools",
"/v1/tools",
]
return [f"{base}{endpoint}" for endpoint in endpoints]
def _endpoint_for_tool(server: dict[str, Any], raw_tool: dict[str, Any]) -> str:
if raw_tool.get("endpoint"):
return str(raw_tool["endpoint"])
protocol = str(raw_tool.get("protocol") or server.get("protocol") or "legacy_http")
if protocol in {"legacy_http", "framework_http", "fastmcp_http"}:
return str(server.get("invoke_endpoint") or "/tools/call")
name = _tool_name(raw_tool) or ""
return str(server.get("invoke_endpoint") or f"/tools/{name}")
def normalize_discovered_tool(server_id: str, server: dict[str, Any], raw_tool: dict[str, Any]) -> tuple[str, dict[str, Any]] | None:
name = _tool_name(raw_tool)
if not name:
return None
discovery = config.get("discovery") or {}
defaults = discovery.get("tool_defaults") or {}
tool_cfg: dict[str, Any] = {
"version": str(raw_tool.get("version") or defaults.get("version") or "1.0.0"),
"server": server_id,
"endpoint": _endpoint_for_tool(server, raw_tool),
"protocol": raw_tool.get("protocol") or server.get("protocol") or defaults.get("protocol") or "legacy_http",
"enabled": bool(raw_tool.get("enabled", defaults.get("enabled", True))),
"idempotent": bool(raw_tool.get("idempotent", defaults.get("idempotent", True))),
"cache_ttl_seconds": int(raw_tool.get("cache_ttl_seconds", defaults.get("cache_ttl_seconds", 0)) or 0),
"timeout_seconds": int(raw_tool.get("timeout_seconds", server.get("timeout_seconds", defaults.get("timeout_seconds", 30))) or 30),
"retry": raw_tool.get("retry") or defaults.get("retry") or {"enabled": False},
"allowed_agents": raw_tool.get("allowed_agents") or defaults.get("allowed_agents") or [],
"allowed_channels": raw_tool.get("allowed_channels") or defaults.get("allowed_channels") or [],
"required_business_keys": raw_tool.get("required_business_keys") or defaults.get("required_business_keys") or [],
"description": raw_tool.get("description") or raw_tool.get("doc") or "",
"input_schema": _tool_schema(raw_tool),
"source": "discovery",
}
return name, tool_cfg
async def discover_server(server_id: str, server: dict[str, Any]) -> tuple[list[tuple[str, dict[str, Any]]], list[dict[str, Any]]]:
errors: list[dict[str, Any]] = []
if not server.get("enabled", True):
return [], []
if not server.get("discover", False):
return [], []
timeout = int((config.get("discovery") or {}).get("timeout_seconds", server.get("timeout_seconds", 10)) or 10)
for url in _catalog_urls(server_id, server):
try:
catalog = await get_json(url, timeout=timeout)
raw_tools = _extract_tools_from_catalog(catalog)
normalized = []
for raw in raw_tools:
item = normalize_discovered_tool(server_id, server, raw)
if item:
normalized.append(item)
if normalized:
return normalized, errors
errors.append({"server": server_id, "url": url, "error": "catalog returned no tools"})
except Exception as exc:
errors.append({"server": server_id, "url": url, "error": str(exc)})
return [], errors
async def sync_discovery() -> DiscoveryResult:
discovered_tools.clear()
errors: list[dict[str, Any]] = []
tools_count = 0
scanned = 0
for server_id, server in (config.get("servers") or {}).items():
if not isinstance(server, dict) or not server.get("discover", False):
continue
scanned += 1
normalized, server_errors = await discover_server(server_id, server)
errors.extend(server_errors)
for name, tool_cfg in normalized:
discovered_tools[name] = tool_cfg
tools_count += 1
discovery_state.update({
"last_sync": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"errors": errors,
"tools": sorted(discovered_tools.keys()),
})
audit("mcp.discovery.completed", {"servers_scanned": scanned, "tools_discovered": tools_count, "errors": len(errors)})
return DiscoveryResult(ok=not errors, servers_scanned=scanned, tools_discovered=tools_count, errors=errors, tools=sorted(discovered_tools.keys()))
@app.on_event("startup")
async def startup_discovery() -> None:
discovery = config.get("discovery") or {}
if discovery.get("enabled", False) and discovery.get("sync_on_startup", True):
try:
await sync_discovery()
except Exception as exc:
audit("mcp.discovery.failed", {"error": str(exc)})
@app.get("/health")
async def health():
return {"status": "ok", "service": "mcp_gateway"}
return {"status": "ok", "service": "mcp_gateway", "version": "1.1.0"}
@app.get("/ready")
async def ready():
enabled_tools = [k for k, v in (config.get("tools") or {}).items() if v.get("enabled", True)]
return {"status": "ready", "tools": enabled_tools}
enabled_tools = [k for k, v in all_tools().items() if v.get("enabled", True)]
return {"status": "ready", "tools": enabled_tools, "discovery": discovery_state}
@app.get("/v1/tools")
async def tools():
return {"tools": [{"name": name, **cfg} for name, cfg in (config.get("tools") or {}).items()]}
return {"tools": [{"name": name, **cfg} for name, cfg in all_tools().items()]}
@app.get("/v1/tools/{tool_name}")
async def tool_detail(tool_name: str):
tool = (config.get("tools") or {}).get(tool_name)
tool = all_tools().get(tool_name)
if not tool:
raise HTTPException(status_code=404, detail=f"Tool not found: {tool_name}")
return {"name": tool_name, **tool}
@app.get("/v1/discovery/servers")
async def discovery_servers():
servers = []
for server_id, server in (config.get("servers") or {}).items():
servers.append({
"id": server_id,
"enabled": server.get("enabled", True),
"discover": server.get("discover", False),
"url": server.get("url"),
"manifest_url": server.get("manifest_url"),
"catalog_endpoint": server.get("catalog_endpoint"),
})
return {"servers": servers, "state": discovery_state}
@app.post("/v1/discovery/sync", response_model=DiscoveryResult)
async def discovery_sync(authorization: str | None = Header(default=None)):
auth_check(authorization)
return await sync_discovery()
@app.post("/v1/tools/{tool_name}/invoke", response_model=ToolResult)
async def invoke(tool_name: str, invocation: ToolInvocation, authorization: str | None = Header(default=None)):
started = time.perf_counter()
auth_check(authorization)
tool = (config.get("tools") or {}).get(tool_name)
tool = all_tools().get(tool_name)
if not tool or not tool.get("enabled", True):
raise HTTPException(status_code=404, detail=f"Tool not found or disabled: {tool_name}")
@@ -217,7 +407,7 @@ async def invoke(tool_name: str, invocation: ToolInvocation, authorization: str
error=error,
cache={"hit": False, "key": ck, "ttl_seconds": ttl},
latency_ms=latency_ms,
metadata={"server": tool.get("server"), **server_metadata},
metadata={"server": tool.get("server"), "source": tool.get("source", "static"), **server_metadata},
)
except Exception as exc:
latency_ms = int((time.perf_counter() - started) * 1000)
@@ -228,5 +418,5 @@ async def invoke(tool_name: str, invocation: ToolInvocation, authorization: str
ok=False,
error=str(exc),
latency_ms=latency_ms,
metadata={"server": tool.get("server")},
metadata={"server": tool.get("server"), "source": tool.get("source", "static")},
)

44
apps/mcp_gateway/config/mcp_gateway.yaml
View File

@@ -1,9 +1,36 @@
# Dedicated MCP Gateway configuration.
# The agent backend/framework calls this gateway; this gateway calls the final MCP servers.
# Discovery allows the gateway to sync tool catalogs from registered MCP servers.
# Static tools below remain supported and override discovered tools with the same name.
discovery:
enabled: true
sync_on_startup: true
timeout_seconds: 10
default_catalog_endpoints:
- /.well-known/mcp-server.json
- /manifest
- /mcp/tools
- /tools/list
- /tools
- /v1/tools
tool_defaults:
version: 1.0.0
protocol: legacy_http
enabled: true
idempotent: true
cache_ttl_seconds: 300
timeout_seconds: 30
retry: {enabled: true, max_attempts: 2, backoff_ms: 250}
allowed_agents: []
allowed_channels: []
required_business_keys: []
servers:
telecom:
enabled: true
discover: false
protocol: legacy_http
transport: http
# Local run: uvicorn mcp.servers.telecom_mcp_server.main:app --port 8100
@@ -12,6 +39,7 @@ servers:
retail:
enabled: true
discover: false
protocol: legacy_http
transport: http
# Local run: uvicorn mcp.servers.retail_mcp_server.main:app --port 8200
@@ -170,3 +198,19 @@ auth:
static_tokens:
runtime-local-token:
agents: []
# Example of a dynamically discovered MCP server.
# Uncomment and adjust the URL to plug a new server that exposes a catalog/manifest.
# servers:
# nf_items:
# enabled: true
# discover: true
# protocol: legacy_http
# transport: http
# url: http://localhost:8400/mcp
# # Optional. If omitted, the gateway tries the discovery.default_catalog_endpoints.
# # manifest_url: http://localhost:8400/.well-known/mcp-server.json
# # catalog_endpoint: /tools
# # invoke_endpoint: /tools/call
# timeout_seconds: 30