Skip to content

server.py and cli.py

Section 7 · From docs/MCP_INTERNALS.md on GitHub

create_layer_from_options loads the config (config or config_path), takes the default registry or the one given, applies read_only_registry when asked, builds the middleware stack (audit if a path is given, metrics unless metrics=False, concurrency if a limit is given, cache if a TTL is given, then any extra middleware), constructs the layer, attaches the cache to change events, registers cloudg://metrics when metrics are on and the URI is free, and loads each --dataset. _load_dataset deliberately bypasses check_path (the operator chose the file), using state.load_dataset_file and ws.unique_name, and raises FileNotFoundError for missing paths.

resolve_flavor turns auto into sdk or native. validate_serve_options(flavor, transport, *, cors_origins, json_response, stateless) resolves the flavor and raises ValueError for combinations a flavor cannot honor (CORS outside the native flavor; --json-response or --stateless with the fastmcp flavor on sse); the CLI calls it before building the layer and turns the error into a usage error. serve_async calls it again, then _ensure_stderr_logging (sets the cloudg logger's level; adds a flagged stderr handler to cloudg only when neither root nor cloudg has any handler), builds token auth, logs that tokens are ignored on stdio, and delegates through _dispatch_serve. Its finally block calls layer.policy.save_vault() and logs where the vault went, so a cancelled or crashed server still persists pseudonyms when the policy has a vault path.

  • _serve_native: builds NativeMCPServer; stdio via run_stdio_async, HTTP via NativeHTTPServer with an HTTPConfig from the options (enable_sse for sse); prints the URL to stderr and resolves ready. The HTTP server builds its own OriginHostGuard.
  • _serve_sdk: build_server(layer, "lowlevel", principal_resolver=...). stdio: on 2.x the SDK's stdio_server(), on 1.x the same inside protected_stdout() with an anyio.wrap_file text wrapper. HTTP: _security_settings() returns TransportSecuritySettings(enable_dns_rebinding_protection=False) (or None before mcp 1.10), because the SDK can only check Host and Origin together and would have to drop the Origin check on a non-loopback bind; on 2.x server.streamable_http_app(...), on 1.x a StreamableHTTPSessionManager behind a Starlette route with a lifespan that runs it; for sse, SseServerTransport("/messages/") routes in front; everything wrapped in TokenAuthASGIMiddleware(app, auth, _guard(host, allowed_origins, allowed_hosts)) and run with uvicorn (lifespan="on").
  • _serve_fastmcp: build_server(layer, "fastmcp"); stdio via run_async(transport="stdio", show_banner=False) (falling back for versions without show_banner); HTTP via server.http_app(transport="http" or "sse", ...). For http it passes path, and json_response / stateless_http when set, after checking with inspect.signature that the installed fastmcp accepts them; when http_app takes host_origin_protection it passes False, since cloudg's guard runs in front. The app is wrapped in TokenAuthASGIMiddleware with the same guard and run with uvicorn.

TokenAuthASGIMiddleware(app, auth, guard=None) passes non-HTTP scopes through; on HTTP it runs guard.check(host, origin) (421 / 403 as JSON-RPC error bodies), then the bearer check (401 with WWW-Authenticate), then stores the principal under "cloudg.principal" in the scope and in scope["state"].

_asgi_endpoint(fn) wraps a raw ASGI callable in a class instance so Starlette's Route treats it as an ASGI app (a plain function would be called as a request handler).

cli.py: _layer_options and _principal_options attach the shared options; _build_layer calls create_layer_from_options with the root command's CloudGConfig (from ctx.obj) unless -c was given on the subcommand, and turns FileNotFoundError / ValueError (an invalid prefix included) into click errors. serve_cmd runs validate_serve_options and checks token specs before building anything. _load_registry imports MODULE:ATTR (attribute path may be dotted; default default_registry), calls it when it is callable and not a Registry, and checks the type. _stdout_console prints tables at a fixed width of 160 columns when stdout is not a TTY, so piped output does not wrap URIs. client_config and _server_command build the config snippets. main() enables logging.captureWarnings and runs the group.

The root cloudg CLI (cloudg/cli.py) adds mcp_group with one add_command line and, when the invoked subcommand is mcp, switches its Rich console to stderr before printing the banner.