diff options
Diffstat (limited to 'src/ollama_web/routes/mcp.py')
| -rw-r--r-- | src/ollama_web/routes/mcp.py | 119 |
1 files changed, 119 insertions, 0 deletions
diff --git a/src/ollama_web/routes/mcp.py b/src/ollama_web/routes/mcp.py new file mode 100644 index 0000000..85d4bb3 --- /dev/null +++ b/src/ollama_web/routes/mcp.py @@ -0,0 +1,119 @@ +"""MCP server configuration API endpoints.""" + +from __future__ import annotations + +import logging +from typing import Any + +from starlette.requests import Request +from starlette.responses import JSONResponse + +from ..mcp import load_mcp_config, save_mcp_config + +logger = logging.getLogger("ollama_web.mcp_route") + + +_ALLOWED_STDIO_FIELDS = {"command", "args", "env", "cwd", "encoding", "encoding_error_handler"} +_ALLOWED_HTTP_FIELDS = {"url", "headers", "timeout", "sse_read_timeout", "terminate_on_close"} + + +async def get_servers(request: Request) -> JSONResponse: + """GET /api/mcp/servers: return the current MCP server configuration.""" + data_dir = request.app.state.settings.data_dir + config = load_mcp_config(data_dir) + return JSONResponse(config) + + +async def put_servers(request: Request) -> JSONResponse: + """PUT /api/mcp/servers: persist a new MCP server configuration. + + The body must be a JSON object with a ``mcpServers`` key. Each server is + validated to contain only recognized fields and a single transport. + """ + try: + body = await request.json() + except Exception as exc: # noqa: BLE001 + logger.warning("invalid MCP config JSON: %s", exc) + return JSONResponse({"error": "Invalid JSON body"}, status_code=400) + + if not isinstance(body, dict) or "mcpServers" not in body: + return JSONResponse( + {"error": "Body must contain a 'mcpServers' object"}, + status_code=400, + ) + + raw_servers = body["mcpServers"] + if not isinstance(raw_servers, dict): + return JSONResponse( + {"error": "'mcpServers' must be an object"}, + status_code=400, + ) + + cleaned: dict[str, Any] = {"mcpServers": {}} + errors: list[str] = [] + for name, server in raw_servers.items(): + if not isinstance(server, dict): + errors.append(f"Server '{name}' must be an object") + continue + + if "command" in server: + valid = _allowed_fields(server, _ALLOWED_STDIO_FIELDS, name, errors) + if valid: + cleaned["mcpServers"][name] = _clean_stdio_server(server) + elif "url" in server: + valid = _allowed_fields(server, _ALLOWED_HTTP_FIELDS, name, errors) + if valid: + cleaned["mcpServers"][name] = _clean_http_server(server) + else: + errors.append( + f"Server '{name}' must define either 'command' (stdio) or 'url' (http)" + ) + + if errors: + return JSONResponse({"error": errors}, status_code=400) + + data_dir = request.app.state.settings.data_dir + save_mcp_config(data_dir, cleaned) + return JSONResponse(cleaned) + + +def _allowed_fields( + server: dict[str, Any], + allowed: set[str], + name: str, + errors: list[str], +) -> bool: + unknown = set(server.keys()) - allowed + if unknown: + errors.append(f"Server '{name}' has unknown fields: {', '.join(sorted(unknown))}") + return False + return True + + +def _clean_stdio_server(server: dict[str, Any]) -> dict[str, Any]: + out: dict[str, Any] = {"command": str(server["command"])} + if "args" in server: + out["args"] = [str(a) for a in server["args"]] + if "env" in server: + out["env"] = {str(k): str(v) for k, v in server["env"].items()} + if "cwd" in server and server["cwd"]: + out["cwd"] = str(server["cwd"]) + for key in ("encoding", "encoding_error_handler"): + if key in server: + out[key] = str(server[key]) + return out + + +def _clean_http_server(server: dict[str, Any]) -> dict[str, Any]: + out: dict[str, Any] = {"url": str(server["url"])} + if "headers" in server: + out["headers"] = {str(k): str(v) for k, v in server["headers"].items()} + for key in ("timeout", "sse_read_timeout"): + if key in server: + try: + out[key] = float(server[key]) + except (TypeError, ValueError): + pass + if "terminate_on_close" in server: + out["terminate_on_close"] = bool(server["terminate_on_close"]) + return out
\ No newline at end of file |