Inside the layer¶

Section 4 · From docs/MCP_INTERNALS.md on GitHub

4.1 Construction¶

CloudGMCPLayer.__init__ (layer.py):

Builds Workspace(config or CloudGConfig()) when no workspace is given.
Loads the policy with Policy.load(policy) unless a Policy instance was passed.
Takes default_registry() when no registry is given and runs _filter_registry. Without filters the registry object is used as is; with any filter, a new Registry receives the specs that pass. Category filters apply to every kind; tool filters only to tools.
Validates the prefix against ^[A-Za-z0-9_.-]{0,64}$ (ValueError otherwise) and stores it with the middleware list, timeout, output cap and identity.
Hooks workspace.on_change(self.notify_change) when the workspace has on_change. That is how dataset changes reach the adapters.

internal_name(exposed) strips the prefix. With a prefix set, a name without it maps to a key that can never be in the registry, so only prefixed names resolve, in-process as well as through the adapters. register_into(prefix=...) assigns layer.prefix directly and skips the constructor's validation (section 11).

4.2 The middleware chain¶

async def _run_chain(self, info, terminal):
    chain = terminal
    for mw in reversed(self.middleware):
        chain = _bind(mw, chain)        # async def call(info): return await mw(info, nxt)
    return await chain(info)

The chain is rebuilt for each call, so layer.use(mw) takes effect on the next request. middleware[0] is outermost. The terminals are _tool_terminal, _resource_terminal, _prompt_terminal and _completion_terminal.

4.3 Running handlers: loop or thread¶

async def _invoke(self, fn, ctx, kwargs):
    self._remember_loop()                 # self._loop = asyncio.get_running_loop()
    ctx.loop = self._loop
    if asyncio.iscoroutinefunction(fn):
        return await fn(ctx, **kwargs)
    result = await asyncio.to_thread(fn, ctx, **kwargs)
    return await maybe_await(result)

Coroutine functions run on the loop and can be cancelled. Plain functions run in the default thread pool executor, so CPU-heavy graph work does not block other requests, at the price that they cannot be cancelled. A sync function that returns an awaitable has it awaited on the loop. Resource, prompt and completion handlers go through _invoke as well.

_remember_loop is what lets the rest of the layer talk to the loop from worker threads: it records the serving loop on the layer and on the context the first time a handler runs.

4.4 Timeouts¶

In _tool_terminal:

timeout = spec.timeout_seconds or self.default_timeout
coro = self._invoke(spec.handler, info.context, kwargs)
raw = await (asyncio.wait_for(coro, timeout) if timeout else coro)

asyncio.TimeoutError becomes _error_result("Tool X timed out", ..., "timeout"). For an async handler, wait_for cancels the coroutine. For a sync handler it cancels the to_thread future, and the thread runs on. Resources, prompts and completions have no timeout. A per-tool timeout_seconds always beats default_timeout, larger or smaller.

4.5 `notify_change` from any thread¶

def notify_change(self, kind, uri=None):
    loop = self._loop
    if loop is not None and loop.is_running():
        try:
            running = asyncio.get_running_loop()
        except RuntimeError:
            running = None
        if running is not loop:
            loop.call_soon_threadsafe(self._dispatch_change, kind, uri)
            return
    self._dispatch_change(kind, uri)

Sync tool handlers call workspace methods in a worker thread, and the workspace fires notify_change from that thread. Adapter listeners touch sessions and asyncio queues, which are not thread-safe, so the layer hops onto the serving loop first. Before any handler has run (_loop is None) the listeners run inline. Listener exceptions are logged at DEBUG and swallowed. ChangeFanout and NativeMCPServer do a second hop of their own because they may be bound to a different loop than the one the layer remembered.

4.6 `report_progress_sync`¶

ToolContext.report_progress is a coroutine and enforces the MCP rule that progress must increase (it remembers the last value and drops anything not greater). report_progress_sync schedules it on ctx.loop with asyncio.run_coroutine_threadsafe and returns without waiting; when there is no callback, no loop, or the loop is closed, it does nothing. The adapters' progress callbacks repeat the monotonic check, so a value is never sent twice even if two code paths report it.

ToolContext.log maps the MCP level to a stdlib level (notice 25, alert 55, emergency 60), logs locally on cloudg.mcp, then calls the adapter's log callback. Unknown levels become info. Callback failures never break the tool.

4.7 Where transforms are applied¶

_transform(value, spec, principal, kind) gets policy.output_pipeline(spec, principal), returns the value unchanged when the pipeline is empty, and otherwise runs it with a fresh TransformContext(direction="output", vault=policy.vault), returning (value, report). _untransform_arguments does the same with input_pipeline and direction="input".

Point	Code	Transformed
Tool arguments	`_tool_terminal`	input pipeline, after `check_call`, before validation; a refusal here calls `policy.record_rejection`
Template variables	`_resource_terminal`	input pipeline
Prompt arguments	`_prompt_terminal`	input pipeline, then filtered to declared argument names
Completion partial value and context arguments	`_completion_terminal`	input pipeline over `{"partial", "arguments"}`
Plain tool result	`_finalise_tool_result`	output pipeline over the structured dict; the text block is a dump of the transformed dict
Handler-built `ToolResult`	`_finalise_tool_result`	`structured`, every `TextContent`, every `ResourceLink`, the text of every `EmbeddedResource` holding `TextResourceContents`, and `meta`
`ctx.link` links	`_transform_links`	`uri`, `name`, `title`, `description` as one dict
Error results	`_error_result`	the message and `data`
Resource contents	`_resource_terminal`	`str` results, JSON results, `TextResourceContents` in a returned list; not `bytes`
Prompt messages	`_prompt_terminal`	`TextContent`, text `EmbeddedResource`, `ResourceLink`
Completion values	`_completion_terminal`	each value wrapped as `{argument_name: value}`, so key-based detectors (names, account ids) see a key, then unwrapped
List definitions	`tools_wire`	not transformed; `outputSchema` withheld when the output pipeline is non-empty

Reports are merged with _merge_report (numbers add up, other values keep the first seen) and stored under _meta["cloudg/transforms"] for tools and plain resource results. The report on list-of-contents resource results, on prompts and on completions is discarded. An ImageContent, or an EmbeddedResource holding a blob, in a handler-built ToolResult is not transformed. When the text block of a plain result is cut, report["truncated_chars"] is the number of characters removed.

4.8 Error results¶

call_tool wraps _run_chain in except MCPLayerError and converts the exception with _error_result(exc.message, spec, principal, exc.code, exc.data). The terminal re-raises MCPLayerError and turns asyncio.TimeoutError and any other exception into error results itself, logging the traceback with logger.exception("Tool %s failed", name). Only get_tool, which runs before the chain, can raise out of call_tool.

get_tool, _resolve_resource and get_prompt call _record_hidden(kind, name, principal) before raising NotFoundError, which forwards to policy.record_hidden when the policy has it; the policy writes a not_found audit entry. Failures there are logged at DEBUG and never change the answer.

read_resource and get_prompt do not catch anything, so policy errors, handler NotFoundErrors and unexpected exceptions propagate to the adapter, which maps them to JSON-RPC errors. get_prompt checks required arguments before building the context.

4.9 Completions¶

async def complete(self, ref, argument, *, principal=None, context_arguments=None) -> dict

Find the function: for ref/prompt, the matching PromptArgument.completion (the prompt name goes through internal_name); for ref/resource, template.completions[argument name], where ref["uri"] must be the template string exactly.
No spec, no function, or the spec not visible to the principal: return empty values without touching the chain.
Build a context with kind="completion", stash the function in ctx.meta["completion_fn"], and run the middleware chain with CallInfo("completion", name, spec, {"argument", "value", "context"}, principal, ctx), where name is the prompt name or the template string. Audit and metrics see the call; CachingMiddleware never caches it.
_completion_terminal calls policy.check_call(spec, principal, {"__completion__": name}), which applies access and rate limits and records the decision with kind completion. It raises, and nothing catches it, so a denied or rate-limited completion becomes a protocol error in the adapters.
Reverse pseudonyms in the partial value and in the context arguments; the restored context arguments go to ctx.meta["arguments"].
Run the function through _invoke with partial=.... Any exception returns empty values (logged at DEBUG).
Run the output pipeline over the values (wrapped under the argument name, 4.7), return the first 100 with total and hasMore.

The function computes every candidate before the cap applies, so keep completion functions cheap or cap them yourself.