In-process use¶

Section 10 · From docs/MCP.md on GitHub

10.1 `CloudGMCPLayer`¶

from cloudg.mcp import CloudGMCPLayer

CloudGMCPLayer(
    config=None, *, workspace=None, policy=None, registry=None,
    include_categories=None, exclude_categories=None, include_tools=None, exclude_tools=None,
    prefix="", middleware=(), default_timeout=300.0, max_output_chars=200_000,
    name="cloudg", version=None, instructions=None,
)

Argument	Default	Meaning
`config`	`CloudGConfig()`	cloudg configuration; used only to build the default workspace
`workspace`	`Workspace(config)`	Shared state (section 11). Pass your own to set allowed roots or share datasets between layers
`policy`	`$CLOUDG_MCP_POLICY` or `standard`	A `Policy`, a profile name, a path, inline JSON or a dict
`registry`	`default_registry()`	Primitives to expose
`include_categories` / `exclude_categories`	none	Filter every kind of primitive by category
`include_tools` / `exclude_tools`	none	Filter tools by internal name
`prefix`	`""`	Prepended to tool and prompt names; validated (at most 64 of `A-Za-z0-9_.-`), `ValueError` otherwise
`middleware`	`()`	Middleware list, outermost first
`default_timeout`	300.0	Seconds; `None` or `0` disables. A tool's own `timeout_seconds` wins
`max_output_chars`	200,000	Cap on the text block of one tool result
`name`, `version`, `instructions`	`"cloudg"`, cloudg's version, built-in text	Server identity and the instructions sent on `initialize` / `server/discover`

Every keyword argument except config is also a field of LayerOptions (cloudg/mcp/layer_options.py, importable from cloudg.mcp.layer), so a configuration can be built once and passed whole: CloudGMCPLayer(config, options=LayerOptions(policy="strict", prefix="cloudg_")). Keyword arguments given next to options override its fields, and an unknown keyword raises TypeError like a normal signature would.

When any filter is given, the layer builds a new Registry with the matching specs, so the registry you pass is not modified. Without filters the layer uses your registry object as is, and later registry.add(...) calls show up in the layer.

Other attributes and methods:

Member	Purpose
`layer.workspace`, `layer.policy`, `layer.registry`, `layer.middleware`	The parts
`layer.use(mw)`	Append a middleware (innermost so far); returns the layer
`layer.on_change(fn)`	Register `fn(kind, uri)` for change events
`layer.notify_change(kind, uri=None)`	Fire a change event; safe from worker threads
`layer.exposed_name(name)` / `layer.internal_name(exposed)`	Add or strip the prefix
`layer.context(principal=None, **fields)`	Build a `ToolContext`
`layer.list_tools(p)`, `list_resources(p)`, `list_resource_templates(p)`, `list_prompts(p)`	Specs visible to principal `p`
`layer.tools_wire(p)`, `resources_wire(p)`, `resource_templates_wire(p)`, `prompts_wire(p)`	The same as wire dicts
`layer.get_tool(name, p)`	The spec, or `NotFoundError`
`await layer.call_tool(name, arguments=None, *, principal=None, context=None)`	`ToolResult`
`await layer.read_resource(uri, *, principal=None, context=None)`	list of `TextResourceContents` / `BlobResourceContents`
`await layer.get_prompt(name, arguments=None, *, principal=None, context=None)`	`PromptResult`
`await layer.complete(ref, argument, *, principal=None, context_arguments=None)`	`{"values", "total", "hasMore"}`
`layer.register_into(server, **kw)`	Section 8
`layer.serve(transport="stdio", **kw)`	Section 10.3

If you pass neither principal nor context, the call runs as Principal.local(). If you pass only a context, its principal is used.

10.2 Calling things¶

import asyncio
from pathlib import Path

from cloudg.config import CloudGConfig
from cloudg.mcp import CloudGMCPLayer, Principal, Workspace

ROOT = Path("docest")


async def main():
    ws = Workspace(CloudGConfig(), allowed_roots=[ROOT], output_dir=ROOT / "out")
    layer = CloudGMCPLayer(policy="standard", workspace=ws)

    res = await layer.call_tool("load_dataset", {"path": "inventory", "name": "prod"})
    print(res.structured["total_assets"], [c.to_wire()["type"] for c in res.content])

    res = await layer.call_tool("get_asset", {"ref": "bastion"})
    print(res.structured["max_severity"], res.meta["cloudg/duration_ms"])

    contents = await layer.read_resource("cloudg://datasets/prod/summary")
    print(contents[0].mime_type, len(contents[0].text))

    prompt = await layer.get_prompt("investigate_asset", {"ref": "bastion"})
    print([m.content.to_wire()["type"] for m in prompt.messages])

    done = await layer.complete({"type": "ref/resource", "uri": "cloudg://assets/{+ref}"},
                                {"name": "ref", "value": "web"})
    print(done)

    analyst = Principal(id="alice", roles={"analyst"})
    res = await layer.call_tool("find_assets", {"internet_exposed": True}, principal=analyst)
    print(res.is_error, res.structured["total"])


asyncio.run(main())

Results from that run (values copied from the printed output):

Call	Result
`load_dataset`	44 assets; content is a `text` block plus a `resource_link` to `cloudg://datasets/prod/summary`
`get_asset("bastion")`	`max_severity` HIGH; links to `cloudg://assets/bastion` and `cloudg://assets/bastion/neighbors`
`read_resource(".../summary")`	one `application/json` text content
`get_prompt("investigate_asset")`	a `text` message with instructions, then a `resource` message embedding the asset's detail
`complete(... "web")`	`{"values": ["web-1", "web-2", "web-acl", "web-alb", "web-tg", "sg-web"], "total": 6, "hasMore": false}`

Completion matches substrings, which is why sg-web is offered for web. Dataset names complete too: {"type": "ref/resource", "uri": "cloudg://datasets/{dataset}/summary"} with an empty value returned {"values": ["prod"], "total": 1, "hasMore": false}, and the framework argument of the compliance_gap_analysis prompt with value CIS returned ["CIS-AWS", "CIS-Azure", "CIS-GCP"].

10.3 `layer.serve`¶

layer.serve("stdio")                                   # blocks until stdin closes
layer.serve("http", host="127.0.0.1", port=8765, auth_tokens=["env:TOKEN:analyst:ci"])

layer.serve(transport, ...) passes everything on to cloudg.mcp.server.serve(layer, transport, ...), which runs serve_async in asyncio.run and returns quietly on Ctrl+C. The async version:

from cloudg.mcp.server import serve_async

await serve_async(
    layer, transport="stdio", *, options=None, flavor="auto", host="127.0.0.1", port=8765,
    path="/mcp", auth=None, auth_tokens=None, allowed_origins=None, allowed_hosts=None,
    cors_origins=(), json_response=False, stateless=False, page_size=100,
    principal_resolver=None, log_level="INFO", ready=None,
)

The arguments mirror the CLI options. They are the fields of cloudg.mcp.server.ServeOptions, so serve_async(layer, "http", options=ServeOptions(port=9000, auth_tokens=[...])) works too; keyword arguments override the fields of options, and an unknown keyword raises TypeError. auth and auth_tokens are left out of the options' repr(), so logging the options cannot leak a token. auth takes a ready TokenAuth; auth_tokens takes spec strings. serve_async leaves the application's logging alone: it sets the cloudg logger to log_level and adds a stderr handler to cloudg only when no handler is configured anywhere. Invalid option combinations (section 5.3) raise ValueError before anything starts. When serving ends, normally or by cancellation, it calls layer.policy.save_vault() so pseudonyms issued during the session are persisted for policies with a vault path. ready is an asyncio.Future that the native HTTP server resolves with the bound (host, port) once it listens, which is how tests start a server on port 0:

import asyncio
from cloudg.mcp.server import serve_async


async def main():
    ready = asyncio.get_running_loop().create_future()
    task = asyncio.create_task(serve_async(layer, "http", flavor="native", port=0, ready=ready))
    host, port = await ready
    print(f"listening on http://{host}:{port}/mcp")
    ...                     # talk to it
    task.cancel()

To build a layer the way the CLI does, with its middleware stack, use cloudg.mcp.server.create_layer_from_options(...). It takes config, config_path, policy, datasets, prefix, the four filters, read_only, audit_log, metrics=True, cache_ttl, max_concurrency, registry, middleware (added innermost), default_timeout, workspace and any extra CloudGMCPLayer keyword. read_only_registry(registry) is the filter behind --read-only.

10.4 `Principal`¶

from cloudg.mcp import Principal

Principal(id="anonymous", roles={"default"}, attributes={})
Principal.local()        # id "local", roles {"default", "local"}

roles drive the policy; attributes are free-form and reach handlers and transforms. Principals are plain dataclasses, and the policy only reads them.

10.5 `ToolContext`¶

Every handler receives a ToolContext as its first argument (named ctx or context).

Member	Meaning
`ctx.layer`, `ctx.workspace`, `ctx.config`	The layer, its workspace, the workspace's cloudg config
`ctx.principal`	The caller
`ctx.kind`, `ctx.name`	`tool`, `resource`, `prompt` or `completion`; the tool or prompt name, the URI, or for a completion the name of the argument being completed
`ctx.request_id`	The JSON-RPC id when an adapter supplied it
`ctx.session`	The adapter's session object (SDK session, fastmcp context, native session) or `None`
`ctx.meta`	Free-form per-call dict; completions find the other arguments in `ctx.meta["arguments"]`
`ctx.links`	Resource links collected by `ctx.link`
`await ctx.report_progress(progress, total=None, message=None)`	Progress notification; values that do not increase are dropped, as MCP requires
`ctx.report_progress_sync(progress, total=None, message=None)`	The same from a sync handler in a worker thread
`await ctx.log(level, data, logger_name="cloudg")`	MCP log notification (and the `cloudg.mcp` Python logger). Levels: debug, info, notice, warning, error, critical, alert, emergency; anything else becomes info
`ctx.link(uri, name, *, title=None, description=None, mime_type="application/json", priority=None)`	Append a `resource_link` to the result

To receive progress and logs in-process, give the context callbacks:

async def on_progress(progress, total, message):
    print("progress", progress, total, message)

ctx = layer.context(Principal.local(), progress_callback=on_progress,
                    log_callback=lambda level, data, name: print("log", level, data, name))
res = await layer.call_tool("cloudg_progress_async", {"steps": 2}, context=ctx)

With a test tool that reports each step twice and then logs at notice, the callbacks received (1, 2, 'step 1'), (2, 2, 'step 2') (the repeated values were dropped) and ('notice', {'done': 2}, 'cloudg'). A sync tool that called ctx.report_progress_sync three times delivered (1, 3, 'step 1'), (2, 3, 'step 2') and (3, 3, 'step 3') through the same callback.

In-process use¶

10.1 CloudGMCPLayer¶

10.2 Calling things¶

10.3 layer.serve¶

10.4 Principal¶

10.5 ToolContext¶

10.1 `CloudGMCPLayer`¶

10.3 `layer.serve`¶

10.4 `Principal`¶

10.5 `ToolContext`¶