"""MCP server configuration, status and diagnostics."""
from typing import Annotated
from fastapi import Depends, HTTPException
from navi.api.deps import require_admin
from navi.auth import User
from ._router import router
from .common import _validate_mcp_name, log
@router.get("/mcp/config")
async def admin_get_mcp_config(
user: Annotated[User, Depends(require_admin)],
) -> dict:
"""Return the current MCP server configurations from mcp_servers.d/."""
from navi.mcp.config import load_mcp_servers
configs = load_mcp_servers()
return {name: cfg.model_dump() for name, cfg in configs.items()}
@router.put("/mcp/config")
async def admin_update_mcp_config(
body: dict,
user: Annotated[User, Depends(require_admin)],
) -> dict:
"""Replace all MCP server configurations (bulk update)."""
from navi.mcp.config import McpServerConfig, save_mcp_servers
validated: dict[str, McpServerConfig] = {}
for name, cfg_data in body.items():
_validate_mcp_name(name)
try:
validated[name] = McpServerConfig.model_validate(cfg_data)
except Exception as exc:
raise HTTPException(
status_code=400,
detail=f"Invalid config for server '{name}': {exc}",
) from exc
save_mcp_servers(validated)
log.info("admin.mcp_config_updated", admin_id=user.id)
return {"ok": True}
@router.get("/mcp/config/{server_name}")
async def admin_get_single_mcp_config(
server_name: str,
user: Annotated[User, Depends(require_admin)],
) -> dict:
_validate_mcp_name(server_name)
"""Return the configuration for a single MCP server."""
from navi.mcp.config import load_mcp_servers
configs = load_mcp_servers()
cfg = configs.get(server_name)
if cfg is None:
raise HTTPException(status_code=404, detail=f"MCP server '{server_name}' not found")
return cfg.model_dump()
@router.put("/mcp/config/{server_name}")
async def admin_update_single_mcp_config(
server_name: str,
body: dict,
user: Annotated[User, Depends(require_admin)],
) -> dict:
_validate_mcp_name(server_name)
"""Create or update a single MCP server configuration."""
from navi.mcp.config import McpServerConfig, load_mcp_servers, save_mcp_servers
try:
validated = McpServerConfig.model_validate(body)
except Exception as exc:
raise HTTPException(
status_code=400,
detail=f"Invalid config for server '{server_name}': {exc}",
) from exc
configs = load_mcp_servers()
configs[server_name] = validated
save_mcp_servers(configs)
log.info("admin.mcp_config_updated_single", server=server_name, admin_id=user.id)
return {"ok": True}
@router.delete("/mcp/config/{server_name}")
async def admin_delete_single_mcp_config(
server_name: str,
user: Annotated[User, Depends(require_admin)],
) -> dict:
_validate_mcp_name(server_name)
"""Remove a single MCP server configuration."""
from navi.mcp.config import load_mcp_servers, save_mcp_servers
configs = load_mcp_servers()
if server_name not in configs:
raise HTTPException(status_code=404, detail=f"MCP server '{server_name}' not found")
del configs[server_name]
save_mcp_servers(configs)
log.info("admin.mcp_config_deleted_single", server=server_name, admin_id=user.id)
return {"ok": True}
@router.post("/mcp/{server_name}/reconnect")
async def admin_reconnect_mcp_server(
server_name: str,
user: Annotated[User, Depends(require_admin)],
) -> dict:
"""Disconnect and reconnect a single MCP server, then re-register its tools."""
from navi.api.deps import get_mcp_manager, get_tool_registry
from navi.mcp.client import McpClient
from navi.mcp.config import load_mcp_servers
from navi.mcp.tools import McpTool
mcp_manager = get_mcp_manager()
if mcp_manager is None:
raise HTTPException(status_code=503, detail="MCP manager not initialized")
configs = load_mcp_servers(mcp_manager.config_path)
cfg = configs.get(server_name)
if cfg is None:
raise HTTPException(
status_code=404, detail=f"MCP server '{server_name}' not found in config"
)
# Drop old client
old_client = mcp_manager.clients.pop(server_name, None)
if old_client:
try:
await old_client.disconnect()
except Exception:
pass
# Unregister old tools for this server
from navi.mcp.tools import build_mcp_name
tool_registry = get_tool_registry()
prefix = build_mcp_name(server_name, "")
for name in list(tool_registry._external_names):
if name.startswith(prefix):
tool_registry.unregister_external(name)
# Connect fresh
client = McpClient(server_name, cfg)
try:
await client.connect()
mcp_manager.clients[server_name] = client
except Exception as exc:
log.warning("admin.mcp_reconnect_failed", server=server_name, error=str(exc))
raise HTTPException(status_code=502, detail=f"Reconnect failed: {exc}") from exc
# Re-register tools
try:
tools = await client.list_tools()
for tool in tools:
mcp_tool = McpTool(
server_name=server_name,
tool_name=tool.name,
description=tool.description or "",
parameters=tool.inputSchema,
manager=mcp_manager,
)
tool_registry.register_external(mcp_tool)
except Exception as exc:
log.warning(
"admin.mcp_list_tools_failed_after_reconnect",
server=server_name,
error=str(exc),
)
log.info("admin.mcp_reconnected", server=server_name, admin_id=user.id)
return {"ok": True}
@router.get("/mcp/status")
async def admin_get_mcp_status(
user: Annotated[User, Depends(require_admin)],
) -> dict:
"""Return connection status and tool counts for every configured MCP server."""
from navi.api.deps import get_mcp_manager
from navi.mcp.config import load_mcp_servers
configs = load_mcp_servers()
manager = get_mcp_manager()
servers = []
for name, cfg in configs.items():
client = manager.clients.get(name) if manager else None
status: dict = {
"name": name,
"transport": cfg.transport,
"connected": client.connected if client else False,
"tool_count": 0,
"instructions": None,
"error": None,
}
if client and client.connected:
status["instructions"] = client.instructions
try:
tools = await client.list_tools()
status["tool_count"] = len(tools)
except Exception as exc:
status["error"] = str(exc)
servers.append(status)
return {"servers": servers}
@router.post("/mcp/test")
async def admin_test_mcp_tool(
body: dict,
user: Annotated[User, Depends(require_admin)],
) -> dict:
"""Execute a single MCP tool call for testing/diagnostics."""
from navi.api.deps import get_mcp_manager
server_name = body.get("server_name")
tool_name = body.get("tool_name")
arguments = body.get("arguments", {})
if not server_name or not tool_name:
raise HTTPException(status_code=400, detail="server_name and tool_name are required")
mcp_manager = get_mcp_manager()
if mcp_manager is None:
raise HTTPException(status_code=503, detail="MCP manager not initialized")
try:
# Admin-only endpoint: run with the shared credential, exactly as the
# diagnostic did before BYOK gating existed.
output, is_error = await mcp_manager.call_tool(
server_name, tool_name, arguments, role="admin"
)
except Exception as exc:
raise HTTPException(status_code=502, detail=f"Tool call failed: {exc}") from exc
return {
"server_name": server_name,
"tool_name": tool_name,
"arguments": arguments,
"output": output,
"is_error": is_error,
}
