Middleware

Section 14 · From docs/MCP.md on GitHub

14.1 Signature and order

async def middleware(info: CallInfo, call_next) -> Any:
    ...
    result = await call_next(info)
    ...
    return result

CallInfo fields: kind (tool, resource, prompt, completion), name (tool or prompt name, the URI for resources, the prompt name or template string for completions), spec, arguments (tool or prompt arguments, template variables, or for completions {"argument": name, "value": partial, "context": {...}}, all as received: pseudonyms not yet reversed), principal, context (the ToolContext) and started (time.monotonic() at creation). call_next returns a ToolResult for tools, a list of resource contents for resources, a PromptResult for prompts and the {"values", "total", "hasMore"} dict for completions. Raise an MCPLayerError to reject; for tools the layer turns it into an error result.

Middleware passed to the layer runs outermost first; layer.use(mw) appends one innermost. The CLI builds this stack:

AuditLogMiddleware (if --audit-log)
  MetricsMiddleware (always)
    ConcurrencyLimitMiddleware (if --max-concurrency)
      CachingMiddleware (if --cache-ttl)
        terminal: policy check, input pipeline, validation, handler, output pipeline

Middleware runs before the terminal's policy.check_call and sees results after the output pipeline. So the audit log records transform reports, and cached results are already transformed for the principal they were cached for. CachingMiddleware runs check_call itself before it serves a hit, so a cached answer spends a rate-limit token, gets the same access decision and lands in the policy's audit buffer like a fresh one.

Completions pass through the chain too, so the audit log and metrics include them; the cache never stores them, and the concurrency limiter counts them when its kinds include resource or completion.

14.2 AuditLogMiddleware

from cloudg.mcp.middleware import AuditLogMiddleware

AuditLogMiddleware(path=None, *, stream=None, salt=None, hash_values=True)

Argument	Meaning
path	JSONL file to append to, created with mode 0600 and its directory created if needed
stream	Alternatively an open text stream
neither	Lines go to the cloudg.mcp.audit Python logger at INFO
salt	HMAC key for value hashes; default $CLOUDG_MCP_AUDIT_SALT, else random per process
hash_values	False logs argument keys only

Argument values are never written. Each is reduced to the first 16 hex characters of HMAC-SHA256(salt, canonical JSON of the value), so someone holding the salt can confirm whether a given value was queried without the log disclosing it. A line from the HTTP walk-through (keys are sorted; one line per call):

{"argument_hashes":{"group_by":"8b696647bea14c90"},"argument_keys":["group_by"],"duration_ms":14.11,"event":"mcp.call","kind":"tool","name":"count_assets","outcome":"ok","principal":{"id":"alice","roles":["analyst","default"]},"request_id":3,"sensitivity":"internal","transforms":{"annotations":{"classification":{"sensitivity":"internal"},"provenance":{"category":"inventory","dataset":"prod","generated_at":"2026-10-08T15:15:37+00:00","kind":"tool","name":"count_assets","policy":"standard","principal_roles":["analyst","default"]}},"content_annotations":{"audience":["user","assistant"],"priority":0.45}},"ts":"2026-10-08T15:15:37.292+00:00"}

A completion from the in-process run (the argument name, the partial value and the context are hashed like any other arguments):

{"argument_hashes":{"argument":"4c28dc8c565760ab","context":"395bcd84fe55f989","value":"dad9f4d26c10bf63"},"argument_keys":["argument","context","value"],"duration_ms":0.35,"event":"mcp.call","kind":"completion","name":"cloudg://assets/{+ref}","outcome":"ok","principal":{"id":"local","roles":["default","local"]},"request_id":null,"sensitivity":"confidential","ts":"2026-10-08T15:14:31.950+00:00"}

And a failed resource read:

{"argument_hashes":{"ref":"06d94df112a60014"},"argument_keys":["ref"],"duration_ms":0.67,"error":{"code":-32602,"type":"ReferenceNotFoundError"},"event":"mcp.call","kind":"resource","name":"cloudg://assets/does-not-exist","outcome":"rejected","principal":{"id":"local","roles":["default","local"]},"request_id":null,"sensitivity":"confidential","ts":"2026-10-08T15:14:31.952+00:00"}

Field	Meaning
ts	UTC timestamp, milliseconds
event	Always mcp.call
kind, name	What was called
principal	{"id", "roles"}
request_id	JSON-RPC id when the adapter supplied one
argument_keys, argument_hashes	Sorted keys; keyed hashes of the values
duration_ms	Wall time inside this middleware
outcome	ok, error (an isError result), denied, rate_limited, rejected (other MCPLayerError), cancelled, exception
sensitivity	The spec's sensitivity
error	{"type", "code"} for raised errors, {"code"} for error results
transforms	The transform report, when there is one

Because the arguments are logged before the input pipeline, a pseudonymized argument is hashed as the pseudonym. Calls to unknown or hidden primitives never reach the middleware (the lookup fails first); the policy records them instead, as decision not_found. The policy's trail is separate from this log (see MCP_PRIVACY.md and the privacy_audit_log tool). Call close() on shutdown if you created the middleware with a path.

14.3 MetricsMiddleware and cloudg://metrics

MetricsMiddleware() (alias TimingMiddleware) counts calls, errors and rejections and keeps latency per primitive (kind:name), with p95 over the last 256 calls. snapshot() returns a dict, reset() clears it, and register_metrics_resource(layer_or_registry, metrics, uri="cloudg://metrics", replace=True) serves the snapshot as a resource (category server, sensitivity internal). The CLI always does both. A read after three tool calls and one resource read:

{
  "uptime_seconds": 0.6,
  "calls": 4,
  "errors": 0,
  "rejected": 0,
  "by_primitive": {
    "resource:cloudg://datasets/prod/summary": {"calls": 1, "errors": 0, "rejected": 0, "avg_ms": 0.52, "p95_ms": 0.52, "max_ms": 0.52, "last_call": "2026-10-08T15:14:31+00:00"},
    "tool:get_asset": {"calls": 2, "errors": 0, "rejected": 0, "avg_ms": 2.39, "p95_ms": 4.67, "max_ms": 4.67, "last_call": "2026-10-08T15:14:31+00:00"},
    "tool:load_dataset": {"calls": 1, "errors": 0, "rejected": 0, "avg_ms": 5.43, "p95_ms": 5.43, "max_ms": 5.43, "last_call": "2026-10-08T15:14:31+00:00"}
  }
}

Completions show up as completion:<prompt or template> once any have run. errors counts error results and unexpected exceptions; rejected counts denials, rate limits, other MCPLayerErrors and cancellations.

14.4 CachingMiddleware

CachingMiddleware(ttl=60.0, maxsize=256, *, predicate=None, cache_resources=False)
cache.attach(layer)       # clear the cache on every layer change event
cache.invalidate()        # clear by hand
cache.stats()             # {"size", "hits", "misses", "ttl"}

The rules:

• By default only tools whose annotations say both read-only and idempotent are cached (is_cacheable_tool(spec)). Tools with cloud_access or exec are not idempotent unless they say so, so live lookups are not cached. predicate(info) -> bool replaces that test; cache_resources=True also caches resource reads. Completions are never cached.
• A hit first runs policy.check_call, the step a hit skips in the terminal: the access decision, the rate-limit token and the policy audit entry apply as for a fresh call, and a rejection propagates the same way.
• The key is (principal id, sorted roles, kind, name, canonical JSON of the arguments), so one principal's transformed result is never served to another.
• Error results are not cached.
• Any call to a tool that is not read-only clears the whole cache, and so does every layer change event once attach was called (the CLI attaches it).
• Entries expire after ttl seconds; the least recently used entry goes when maxsize is exceeded.
• Hits are deep copies marked with _meta["cloudg/cache"] = "hit".

A second get_asset(ref="bastion") in the in-process run came back with "cloudg/cache": "hit" and an audit duration_ms of 0.12 against 4.69 for the first call.

14.5 ConcurrencyLimitMiddleware

ConcurrencyLimitMiddleware(limit=8, *, per_principal=None, queue_timeout=30.0, kinds=("tool",))

At most limit calls of the given kinds run at once, and at most per_principal per principal id when set. A call that waits longer than queue_timeout seconds for a slot fails with RateLimitedError("Server busy: too many concurrent MCP calls"), which a tool caller sees as an error result with -31029. queue_timeout=None waits forever. Completions count when kinds contains "resource" or "completion". in_flight reports the current count. --max-concurrency N adds it with limit=N.

14.6 RetryMiddleware

RetryMiddleware(retries=2, *, backoff=0.5, retry_on=(ConnectionError, TimeoutError),
                retry_codes=("timeout",))

Retries read-only, idempotent tools when the call raises one of retry_on or returns an error result whose cloudg/error_code is in retry_codes. The wait doubles each time (0.5 s, 1 s, ...). A result that needed retries gets _meta["cloudg/retries"]. The CLI does not add it. Keep in mind that a timed-out sync handler keeps running in its thread, so retrying timeouts can stack threads doing the same work.

14.7 Writing your own

Middleware is any async callable. This one tags results with a request label and refuses tool calls outside office hours for a role:

import datetime
from cloudg.mcp import AccessDeniedError, ToolResult


async def office_hours(info, call_next):
    if info.kind == "tool" and "contractor" in info.principal.roles:
        hour = datetime.datetime.now().hour
        if not 8 <= hour < 18:
            raise AccessDeniedError("Contractors can call tools between 08:00 and 18:00",
                                    data={"hour": hour})
    result = await call_next(info)
    if isinstance(result, ToolResult):
        result.meta["example.com/request-label"] = f"{info.kind}:{info.name}"
    return result


layer = CloudGMCPLayer(policy="standard", middleware=[office_hours])
# or: layer.use(office_hours)

Run at 20:00 against the test registry's echo tool, a contractor got {"content": [{"type": "text", "text": "Contractors can call tools between 08:00 and 18:00"}], "isError": true, "_meta": {"cloudg/error_code": -31001, "cloudg/error_data": {"hour": 20}}}, and the local user's result carried "example.com/request-label": "tool:echo" in _meta.

Use your own reverse-DNS prefix for _meta keys; prefixes whose second label is mcp or modelcontextprotocol are reserved by the protocol.

Middleware sees arguments before the input pipeline, so do not branch on identifiers in them when a pseudonymizing profile is active. Middleware that needs real identifiers belongs inside the handler, or in a transform (see MCP_PRIVACY.md).