Mounting into an existing MCP server

Section 8 · From docs/MCP.md on GitHub

8.1 register_into

layer.register_into(server, *, prefix=None, include=None, principal_resolver=None,
                    list_changed=True) -> binding
# equivalent: cloudg.mcp.adapters.register_into(layer, server, ...)
#             cloudg.mcp.register_into(layer, server, ...)

Argument	Meaning
server	An mcp.server.mcpserver.MCPServer (mcp 2.x), an mcp.server.fastmcp.FastMCP (mcp 1.x), an mcp.server.lowlevel.Server (1.x or 2.x), a fastmcp.FastMCP (2.x, 3.x, 4.x), or the layer's own NativeMCPServer (returned unchanged)
prefix	Sets the layer's prefix through layer.set_prefix(), which raises ValueError for anything but up to 64 of A-Za-z0-9_.-; tool and prompt names become prefix + name. The prefix belongs to the layer, so mounting one layer twice with different prefixes logs a warning and renames it everywhere: use one layer per prefix. Resource URIs are never prefixed (they already start with cloudg://)
include	Subset of {"tools", "resources", "templates", "prompts", "completions", "logging", "subscriptions"}; default all. Anything else raises ValueError
principal_resolver	fn(RequestInfo) -> Principal or None; None falls back to the default resolver (8.8)
list_changed	Advertise listChanged capabilities on SDK servers so clients act on change notifications

The server kind is detected by cloudg.mcp.adapters.detect_server_kind(server), which returns native, fastmcp, sdk-highlevel or lowlevel, falls back to duck typing for subclasses and wrappers, and raises TypeError for anything it does not recognize.

What every adapter guarantees:

• Tools are listed from the layer's own wire dicts, so inputSchema, outputSchema, annotations, title and _meta reach the client exactly as cloudg built them. The host framework never re-derives a schema from a Python signature.
• The host's own tools, resources and prompts keep working. List results are the host's entries followed by cloudg's; a name or URI the host already defines keeps the host's version and logs a warning.
• Calls for names and URIs cloudg owns go to the layer; everything else goes to the host's original handler.
• structuredContent, isError, resource links and _meta pass through.
• Progress and log notifications reach the client; change notifications (8.10) are forwarded.
• Every request goes through layer.call_tool / read_resource / get_prompt / complete, so policy, transforms and middleware apply.

Mount after the host has registered its own handlers. The low-level adapters wrap whatever handler is installed when register_into runs; a host handler registered later replaces cloudg's wrapper. On mcp 1.30, calling register_into before the host's @server.list_tools() decorator left tools/list returning only the host's one tool.

8.2 mcp 2.x MCPServer

import asyncio
from pathlib import Path

from mcp import Client
from mcp.server.mcpserver import MCPServer

from cloudg.mcp import CloudGMCPLayer, Workspace

ROOT = Path("docest")
server = MCPServer("platform-tools")


@server.tool()
def ticket_count(team: str) -> int:
    """How many open tickets a team has (the host's own tool)."""
    return 7


layer = CloudGMCPLayer(policy="standard", workspace=Workspace(allowed_roots=[ROOT]))
layer.workspace.load(ROOT / "inventory" / "inventory-map.json", "prod")
binding = layer.register_into(server, prefix="cloudg_")
print("binding:", type(binding).__name__, "sdk_major =", binding.sdk_major)


async def main() -> None:
    for mode in ("legacy", "auto"):
        async with Client(server, mode=mode) as client:
            tools = (await client.list_tools()).tools
            names = [t.name for t in tools]
            print(mode, client.protocol_version, len(names), names[:3], "...", names[-2:])
            res = await client.call_tool("cloudg_count_assets", {"group_by": "provider"})
            print("  structured:", res.structured_content["groups"])
            res = await client.call_tool("ticket_count", {"team": "sec"})
            print("  host tool:", res.content[0].text)


asyncio.run(main())

Output (mcp 2.3.0):

binding: LowLevelBinding sdk_major = 2
legacy 2025-11-25 74 ['ticket_count', 'cloudg_workspace_status', 'cloudg_list_datasets'] ... ['cloudg_list_detectors', 'cloudg_privacy_audit_log']
  structured: {'AWS': 36, 'AZURE': 5, 'GCP': 3}
  host tool: 7
auto 2026-07-28 74 ['ticket_count', 'cloudg_workspace_status', 'cloudg_list_datasets'] ... ['cloudg_list_detectors', 'cloudg_privacy_audit_log']
  structured: {'AWS': 36, 'AZURE': 5, 'GCP': 3}
  host tool: 7

To serve it, use the host's own runner (server.run(), server.streamable_http_app(), and so on). On MCPServer, a tool name cloudg does not own falls through to the host, which answers an unknown name with an isError result instead of a JSON-RPC error.

8.3 mcp 1.x FastMCP

import asyncio
from pathlib import Path

from mcp.server.fastmcp import FastMCP
from mcp.shared.memory import create_connected_server_and_client_session

from cloudg.mcp import CloudGMCPLayer, Workspace

ROOT = Path("docest")
server = FastMCP("platform-tools")


@server.tool()
def ticket_count(team: str) -> int:
    """How many open tickets a team has (the host's own tool)."""
    return 7


layer = CloudGMCPLayer(policy="standard", workspace=Workspace(allowed_roots=[ROOT]))
layer.workspace.load(ROOT / "inventory", "prod")
binding = layer.register_into(server, prefix="cloudg_")
print("binding:", type(binding).__name__, "sdk_major =", binding.sdk_major)


async def main() -> None:
    async with create_connected_server_and_client_session(server) as session:
        tools = (await session.list_tools()).tools
        print(len(tools), [t.name for t in tools][:3])
        res = await session.call_tool("cloudg_count_assets", {"group_by": "provider"})
        print("structured:", res.structuredContent["groups"])


asyncio.run(main())

Output (mcp 1.30.0; the SDK's own INFO log lines removed):

binding: LowLevelBinding sdk_major = 1
74 ['ticket_count', 'cloudg_workspace_status', 'cloudg_list_datasets']
structured: {'AWS': 36, 'AZURE': 5, 'GCP': 3}

8.4 The low-level Server

mcp 2.x, a server that already answers tools/list and tools/call itself:

import asyncio
import json
from pathlib import Path

import mcp.types as types
from mcp import Client
from mcp.server.lowlevel import Server

from cloudg.mcp import CloudGMCPLayer, Workspace

ROOT = Path("docest")
HOST_TOOL = types.Tool(name="ping_host", description="Host liveness.",
                       input_schema={"type": "object", "properties": {}})


async def list_tools(ctx, params):
    return types.ListToolsResult(tools=[HOST_TOOL])


async def call_tool(ctx, params):
    return types.CallToolResult(content=[types.TextContent(type="text", text="pong")])


server = Server("platform", on_list_tools=list_tools, on_call_tool=call_tool)
layer = CloudGMCPLayer(policy="standard", workspace=Workspace(allowed_roots=[ROOT]),
                       include_categories=["workspace", "inventory"])
layer.workspace.load(ROOT / "inventory", "prod")
layer.register_into(server, prefix="cloudg_")


async def main() -> None:
    async with Client(server, mode="auto") as client:
        names = [t.name for t in (await client.list_tools()).tools]
        print(len(names), names[:4])
        print((await client.call_tool("ping_host", {})).content[0].text)
        res = await client.call_tool("cloudg_list_regions", {})
        print(json.dumps(res.structured_content)[:160])


asyncio.run(main())

20 ['ping_host', 'cloudg_workspace_status', 'cloudg_list_datasets', 'cloudg_load_dataset']
pong
{"dataset": "prod", "total": 5, "items": [{"region": "us-east-1", "providers": ["AWS"], "assets": 24, "accounts": 1}, {"region": "global", "providers": ["AWS"],

include_categories cut the catalog to 19 tools; with the host's tool that makes 20.

mcp 1.x, with the decorator API:

import mcp.types as types
from mcp.server.lowlevel import Server

server = Server("platform")


@server.list_tools()
async def list_tools() -> list[types.Tool]:
    return [types.Tool(name="ping_host", description="Host liveness.",
                       inputSchema={"type": "object", "properties": {}})]


@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    return [types.TextContent(type="text", text="pong")]


layer = CloudGMCPLayer(policy="standard", workspace=Workspace(allowed_roots=[ROOT]))
layer.workspace.load(ROOT / "inventory", "prod")
layer.register_into(server, prefix="cloudg_")      # after the decorators

With mcp 1.30 the in-memory session listed 74 tools, ping_host answered pong, and cloudg_count_assets returned {'AWS': 36, 'AZURE': 5, 'GCP': 3}.

A low-level server with no handlers of its own simply becomes a cloudg server. That is what build_server(layer, "lowlevel") does (8.9).

8.5 Standalone fastmcp

import asyncio
from pathlib import Path

import fastmcp

from cloudg.mcp import CloudGMCPLayer, Workspace

ROOT = Path("docest")
server = fastmcp.FastMCP("platform-tools")


@server.tool
def ticket_count(team: str) -> int:
    """How many open tickets a team has (the host's own tool)."""
    return 7


layer = CloudGMCPLayer(policy="standard", workspace=Workspace(allowed_roots=[ROOT]))
layer.workspace.load(ROOT / "inventory", "prod")
binding = layer.register_into(server, prefix="cloudg_")


async def main() -> None:
    async with fastmcp.Client(server) as client:
        tools = await client.list_tools()
        print(len(tools), [t.name for t in tools][:3])
        res = await client.call_tool_mcp("cloudg_count_assets", {"group_by": "provider"})
        print("structured:", res.structuredContent["groups"])
        text = await client.read_resource("cloudg://datasets/prod/summary")
        print("resource bytes:", len(text[0].text))


asyncio.run(main())

Output, identical on fastmcp 4.0.11, 3.4.8 and 2.14.7 apart from the version line:

fastmcp 4.0.11 mcp 2.3.0 binding: FastMCPBinding
74 ['ticket_count', 'cloudg_workspace_status', 'cloudg_list_datasets']
structured: {'AWS': 36, 'AZURE': 5, 'GCP': 3}
resource bytes: 1783

fastmcp 4 also warns that structuredContent is deprecated in favor of structured_content on its client result; use whichever your fastmcp version supports.

cloudg registers fastmcp component objects, adds a fastmcp middleware that filters list results through the policy for the calling principal, and binds completions and resource subscriptions on fastmcp's inner SDK server, since fastmcp components cannot express those. fastmcp features such as its own middleware, auth providers, tags and the in-memory client keep working. According to fastmcp 4's upgrade guide, mount(namespace=...) renames tools and rewrites resource URIs, so resource links inside cloudg results would no longer match (not tested here). Use prefix= rather than mounting a cloudg-only fastmcp server under a namespace.

8.6 Inside an existing Starlette or ASGI app

The SDK's streamable_http_app() returns a Starlette app. Two details matter when you mount it: Starlette's Mount does not run a sub-app's lifespan, so start the SDK session manager in the parent app's lifespan; and mounting under /mcp with the SDK's default inner path makes POST /mcp answer with a redirect, so mount at / and set streamable_http_path instead.

from contextlib import asynccontextmanager
from pathlib import Path

from mcp.server.transport_security import TransportSecuritySettings
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Mount, Route

from cloudg.mcp import CloudGMCPLayer, Principal, Workspace
from cloudg.mcp.adapters import build_server
from cloudg.mcp.native.auth import TokenAuth
from cloudg.mcp.server import TokenAuthASGIMiddleware

ROOT = Path("docest")
layer = CloudGMCPLayer(policy="standard", workspace=Workspace(allowed_roots=[ROOT]))
layer.workspace.load(ROOT / "inventory", "prod")


def resolver(info):
    # TokenAuthASGIMiddleware already put the principal on the request scope;
    # add the tenant header as an attribute for policies and audit.
    if info.principal is not None:
        info.principal.attributes["tenant"] = info.header("x-tenant", "unknown")
        return info.principal
    return None  # fall back to the default resolver (anonymous on HTTP)


mcp_server = build_server(layer, "lowlevel", principal_resolver=resolver)
mcp_app = mcp_server.streamable_http_app(
    streamable_http_path="/mcp",
    transport_security=TransportSecuritySettings(
        enable_dns_rebinding_protection=True,
        allowed_hosts=["127.0.0.1:*", "localhost:*"],
        allowed_origins=["http://localhost:*"],
    ),
)


async def health(request):
    return JSONResponse({"ok": True})


@asynccontextmanager
async def lifespan(app):
    # Mount() does not run a sub-app's lifespan: start the session manager here
    async with mcp_server.session_manager.run():
        yield


auth = TokenAuth({"t-analyst": Principal(id="ana", roles={"analyst"})})
app = Starlette(
    routes=[Route("/health", health), Mount("/", app=TokenAuthASGIMiddleware(mcp_app, auth))],
    lifespan=lifespan,
)

Run with uvicorn asgi_app:app --port 8797. Results with mcp 2.3.0:

$ curl -s http://127.0.0.1:8797/health
{"ok":true}
$ curl -si -X POST http://127.0.0.1:8797/mcp -H 'Content-Type: application/json' \
    -H 'Accept: application/json, text/event-stream' -d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{"_meta":{...}}}'
HTTP/1.1 401 Unauthorized
www-authenticate: Bearer realm="cloudg-mcp"
{"jsonrpc":"2.0","id":null,"error":{"code":-32600,"message":"Unauthorized"}}
$ curl -s -X POST http://127.0.0.1:8797/mcp -H 'Authorization: Bearer t-analyst' -H 'X-Tenant: acme' \
    -H 'MCP-Protocol-Version: 2026-07-28' -H 'Mcp-Method: tools/call' -H 'Mcp-Name: count_assets' ... \
    -d '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"count_assets","arguments":{"group_by":"provider"},"_meta":{...}}}'
{"jsonrpc":"2.0","id":2,"result":{"_meta":{"cloudg/transforms":{...,"provenance":{...,"principal_roles":["analyst"]}}},"cloudg/duration_ms":1,"io.modelcontextprotocol/serverInfo":{"name":"cloudg","version":"0.6.0"}},"content":[...

TokenAuthASGIMiddleware(app, auth, guard=None) first runs guard (an OriginHostGuard, section 7) when one is given, answering 421 or 403. It then checks the bearer header on HTTP scopes, answers 401 when auth.required and the token is wrong, and otherwise stores the principal in scope["cloudg.principal"] and scope["state"]["cloudg.principal"]. The adapters' default resolution finds it there. Lifespan and non-HTTP scopes pass through untouched, and with neither auth nor guard the middleware does nothing. The example above keeps the SDK's own TransportSecuritySettings; to get cloudg's rules instead, pass guard=OriginHostGuard("127.0.0.1", allowed_origins=None, allowed_hosts=None) (from cloudg.mcp.native.http) and TransportSecuritySettings(enable_dns_rebinding_protection=False), which is what cloudg mcp serve --flavor sdk does. The example was re-run after that change and behaved as shown.

If your app already authenticates users, skip TokenAuth and put your own Principal into scope["cloudg.principal"] in your middleware, or write a resolver that reads your session.

For the low-level server, mcp_server.streamable_http_app(...) also accepts the SDK's auth=AuthSettings(...) and token_verifier=... for OAuth; the default resolver then turns the validated AccessToken into a principal (8.8). I did not test an OAuth setup for this guide.

8.7 Prefixes and name collisions

prefix exists so that cloudg's 74 tools cannot collide with the host's. With prefix="cloudg_":

• tools/list shows cloudg_find_assets; prompts get the same prefix.
• The adapters only claim names that start with the prefix and whose remainder is a cloudg tool. find_assets without the prefix goes to the host.
• Resource URIs are not prefixed.
• With a prefix set, only prefixed names resolve, in-process too: layer.call_tool("progress_async") on a layer with prefix="cloudg_" raised NotFoundError -32602 "Unknown tool: progress_async", while "cloudg_progress_async" worked.
• The constructor validates the prefix: at most 64 characters from A-Za-z0-9_.-. CloudGMCPLayer(prefix="cloudg:") raises ValueError: Invalid prefix 'cloudg:': use up to 64 of A-Z a-z 0-9 _ . -. Exposed names must also match ^[A-Za-z0-9_.-]{1,128}$.

register_into(prefix=...) goes through layer.set_prefix(), the same check as the constructor, so register_into(layer, server, prefix="bad:") raises ValueError at once. The prefix belongs to the shared layer object, so mounting one layer into two servers with different prefixes leaves both using the last one (a warning is logged). Build one layer per prefix.

8.8 Principal resolvers

The adapters fill a cloudg.mcp.native.auth.RequestInfo for every request:

Field	Content
transport	stdio, http, sse or memory
headers	HTTP headers with lower-cased names; empty on stdio
session_id	Mcp-Session-Id when present
client_info	The client's clientInfo dict when the framework exposes it
access_token	The SDK's AccessToken when the host validated an OAuth bearer token
principal	A principal stored by an auth middleware (scope["cloudg.principal"])
raw	The framework's own request context object

info.header(name, default=None) reads a header case-insensitively, and info.bearer_token returns the token from Authorization: Bearer ... or None.

A resolver is fn(info) -> Principal | None. Returning None, or raising (which is logged), falls back to default_principal_resolver:

1. info.principal, if an auth layer set one.
2. An OAuth access token: id from client_id (or oauth-client), roles {"default", *scopes}, attributes {"auth": "oauth", "scopes": [...]}.
3. The local user (Principal.local()) for stdio and memory transports.
4. Principal(id="anonymous", roles={"default"}) for anything else.

Map OAuth scopes to the role names your policy uses, or keep the scope names as roles. A resolver that trusts a gateway's identity header:

from cloudg.mcp import Principal


def gateway_resolver(info):
    # Only safe when nothing but the gateway can reach this server.
    user = info.header("x-authenticated-user")
    groups = info.header("x-authenticated-groups", "")
    if not user:
        return None
    return Principal(id=user, roles={g.strip() for g in groups.split(",") if g.strip()}
                     or {"default"}, attributes={"auth": "gateway"})


layer.register_into(server, principal_resolver=gateway_resolver)

Never derive roles from clientInfo; any client can send any value there.

8.9 build_server

from cloudg.mcp.adapters import build_server

build_server(layer, flavor="auto", *, name=None, principal_resolver=None, include=None,
             **native_kwargs)

Flavor	Returns
auto	lowlevel when mcp imports, else native
sdk, lowlevel	mcp.server.lowlevel.Server(name or layer.name, version=layer.version, instructions=layer.instructions) with cloudg registered
mcpserver	MCPServer (mcp 2.x) or FastMCP (mcp 1.x) with cloudg registered
fastmcp	fastmcp.FastMCP(name or layer.name, instructions=...) with cloudg registered
native	NativeMCPServer(layer, principal_resolver=..., **native_kwargs)

The SDK and fastmcp objects carry the binding at server._cloudg_binding. The shorter cloudg.mcp.build_server(layer, flavor="auto") takes only the flavor.

Running a built server yourself, for example with the SDK's stdio helpers on mcp 2.x:

import anyio
from mcp.server.stdio import stdio_server

server = build_server(layer, "lowlevel")


async def main():
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

anyio.run(main)

This is what cloudg mcp serve --flavor sdk does on stdio. For everything else, prefer layer.serve(...) (10.3).

8.10 Change notifications

The workspace reports changes to the layer: loading a dataset fires ("resources", None) and ("resource", uri) for cloudg://workspace, cloudg://datasets and the dataset's summary; mutations such as suppressing findings fire ("resource", uri) for those URIs. Tools may also call ctx.layer.notify_change("tools") or ("prompts"). The adapters turn these into protocol notifications:

Event	Handshake era	2026-07-28 era
tools, prompts, resources	notifications/*/list_changed to every session the adapter has seen	event on the subscriptions/listen streams that asked for it
resource with a URI	notifications/resources/updated, only to sessions that called resources/subscribe for that URI	event on listen streams whose resourceSubscriptions include the URI

On mcp 2.x the adapter publishes 2026 events on the server's subscription bus, installing a subscriptions/listen handler when the server has none. Notifications raised from a worker thread (a sync handler) are marshalled onto the event loop first.