Core contracts¶

Section 3 · From docs/MCP_INTERNALS.md on GitHub

3.1 Specs¶

All four specs inherit _BaseSpec:

Field	Type	Default	Notes
`name`	`str`	required	Tools and prompts: wire name before the prefix. Resources: display name
`handler`	callable	required	Sync or async; first parameter is the `ToolContext`
`title`	`str or None`	`None`	Display title
`description`	`str`	`""`	Falls back to the handler's cleaned docstring
`category`	`str`	`"general"`	Used by filters, policies and `_meta`
`tags`	`set[str]`	empty	Policies can match on them
`sensitivity`	`Sensitivity`	`CONFIDENTIAL`	Prompts default to `INTERNAL`
`capabilities`	`set[Capability]`	`{READ_STATE}`	Drives annotations and policy capability denials
`meta`	`dict`	empty	Merged into the wire `_meta` after the cloudg keys
`icons`	`list[Icon] or None`	`None`	`Icon(src, mime_type, sizes, theme)`; prefer `data:` URIs

ToolSpec adds input_schema (derived from the handler when None), output_schema, annotations: ToolAnnotations, transform_hints (per-tool options for the policy's transforms) and timeout_seconds. Its __post_init__ validates the name against ^[A-Za-z0-9_.-]{1,128}$, copies title into annotations.title (older clients only read the latter), derives unset hints (3.5) and builds the input schema.

ResourceSpec adds uri, mime_type (default application/json), annotations: ContentAnnotations and size. ResourceTemplateSpec adds uri_template, mime_type, annotations and completions: {variable: fn(ctx, partial) -> list[str]}, and compiles the template to a regex. PromptSpec adds arguments: list[PromptArgument], where PromptArgument(name, description="", required=False, completion=None, title=None).

Handlers:

Spec	Signature	May return
Tool	`handler(ctx, **arguments)`	dict, list, scalar, pydantic model, dataclass, object with `to_dict()`, or `ToolResult`
Resource	`handler(ctx)`	`str`, `bytes`, JSON-serializable value, or a non-empty list of `TextResourceContents` / `BlobResourceContents`
Template	`handler(ctx, **variables)`	as resources
Prompt	`handler(ctx, **arguments)`	`PromptResult`, list of `PromptMessage`, or `str`
Completion	`fn(ctx, partial)`	iterable of values (converted with `str`)

3.2 `Registry`¶

reg = Registry()
reg.add(spec, *, replace=False)       # ValueError on duplicates unless replace=True
reg.merge(other, *, replace=False)    # add every spec of another registry
reg.find_template(uri)                # (spec, variables) or None
len(reg)                              # total number of primitives
reg.tools / reg.resources / reg.templates / reg.prompts   # dicts keyed by name / uri / template / name

Decorators register the decorated function and return it unchanged:

Decorator	Parameters
`@reg.tool(name=None, *, ...)`	`title`, `description`, `category`, `tags`, `sensitivity`, `capabilities`, `read_only`, `destructive`, `idempotent`, `open_world`, `output_schema`, `transform_hints`, `timeout_seconds`, `icons`
`@reg.resource(uri, *, ...)`	`name`, `title`, `description`, `mime_type`, `category`, `tags`, `sensitivity`, `annotations`, `capabilities`, `icons`
`@reg.resource_template(uri_template, *, ...)`	as `resource`, plus `completions`
`@reg.prompt(name=None, *, ...)`	`title`, `description`, `arguments`, `category`, `tags`, `sensitivity`, `icons`

find_template tries templates longest template string first, so cloudg://assets/{+ref}/neighbors wins over cloudg://assets/{+ref} for .../neighbors URIs.

3.3 Wire shapes¶

ToolSpec.to_wire(name=None, *, include_output_schema=True) for workspace_status under the open profile:

{
  "name": "workspace_status",
  "description": "Show what is loaded: every dataset (asset / edge / finding counts,\nsource, which analyses are cached), the active dataset, and the\ndirectories file tools may read and write. Call this first; if no\ndataset is loaded, use load_dataset or map_inventory.",
  "inputSchema": {"additionalProperties": false, "properties": {}, "type": "object"},
  "outputSchema": {
    "additionalProperties": true,
    "properties": {
      "active_dataset": {"anyOf": [{"type": "string"}, {"type": "null"}], "default": null},
      "datasets": {"items": {"additionalProperties": true, "type": "object"}, "type": "array"},
      "allowed_roots": {"anyOf": [{"items": {"type": "string"}, "type": "array"}, {"type": "null"}], "default": null},
      "output_dir": {"anyOf": [{"type": "string"}, {"type": "null"}], "default": null}
    },
    "type": "object"
  },
  "annotations": {"title": "Workspace status", "readOnlyHint": true, "destructiveHint": false,
                  "idempotentHint": true, "openWorldHint": false},
  "title": "Workspace status",
  "_meta": {"cloudg/category": "workspace", "cloudg/sensitivity": "internal",
            "cloudg/tags": ["start-here"]}
}

ResourceSpec.to_wire():

{"uri": "cloudg://workspace", "name": "workspace_resource", "mimeType": "application/json",
 "description": "Loaded datasets, the active dataset and allowed directories.",
 "title": "Workspace",
 "_meta": {"cloudg/category": "workspace", "cloudg/sensitivity": "internal"}}

ResourceTemplateSpec.to_wire():

{"uriTemplate": "cloudg://assets/{+ref}", "name": "asset_resource", "mimeType": "application/json",
 "description": "One asset of the active dataset (ref = id, ARN or unique name, URL-encoded).",
 "title": "Asset",
 "_meta": {"cloudg/category": "inventory", "cloudg/sensitivity": "confidential"}}

PromptSpec.to_wire(name=None):

{"name": "compliance_gap_analysis",
 "description": "Failing controls, responsible assets and a remediation backlog for one framework.",
 "arguments": [
   {"name": "framework", "required": true, "description": "Framework, e.g. CIS-AWS, PCI-DSS, NIST-800-53."},
   {"name": "dataset", "required": false, "description": "Dataset name; empty = active dataset."}
 ],
 "title": "Compliance gap analysis",
 "_meta": {"cloudg/category": "prompts", "cloudg/sensitivity": "confidential", "cloudg/tags": ["workflow"]}}

ToolResult.to_wire() emits content, isError, and structuredContent and _meta when set. ToolResult.error(message, **meta) builds an error result. The content classes emit type plus their fields, annotations only when non-empty and _meta only when set; ResourceLink omits empty optional fields. Tool and prompt wire dicts always include description, even when it is the empty string.

ToolSpec.to_wire and PromptSpec.to_wire validate the exposed name (after the prefix) against the same name pattern and raise ValueError when it does not match.

3.4 Schema generation¶

input_schema_for(fn):

Builds (and caches on the function as __cloudg_mcp_args_model__) a pydantic model from the signature, skipping a first parameter named ctx or context and any *args or **kwargs. Annotations come from typing.get_type_hints(include_extras=True), so Annotated[int, Field(ge=1, description=...)] carries constraints and descriptions; when forward references cannot be resolved every parameter becomes Any. The model uses extra="forbid".
Calls model_json_schema(), drops pydantic's generated title keys, and inlines local $refs into #/$defs/... (several clients do not follow references). A recursive model keeps its $defs.
Forces type: object, a properties key and additionalProperties: false.

A handler with a nested model:

class Window(BaseModel):
    start: str
    end: str | None = None


def changes(ctx, window: Window, kinds: list[Literal["asset", "edge"]] = ["asset"],
            limit: Annotated[int, Field(ge=1, le=100)] = 10) -> dict:
    """Changes in a window."""

produces:

{
  "additionalProperties": false,
  "properties": {
    "window": {
      "properties": {
        "start": {"type": "string"},
        "end": {"anyOf": [{"type": "string"}, {"type": "null"}], "default": null}
      },
      "required": ["start"],
      "type": "object"
    },
    "kinds": {"default": ["asset"], "items": {"enum": ["asset", "edge"], "type": "string"}, "type": "array"},
    "limit": {"default": 10, "maximum": 100, "minimum": 1, "type": "integer"}
  },
  "required": ["window"],
  "type": "object"
}

validate_arguments(fn, arguments) validates with the same model and returns a dict of the model's attributes:

validate_arguments(changes, {"window": {"start": "2026-01-01"}, "limit": "5"})
# {'window': Window(start='2026-01-01', end=None), 'kinds': ['asset'], 'limit': 5}

Note that nested models arrive as model instances, not dicts, and that "5" was coerced to 5. On failure it raises InvalidArgumentsError("Invalid arguments: loc: msg; ...") with data set to [{"loc", "msg", "type"}, ...]; the input values are not echoed.

output_schema_for(Model) renders an output schema the same way, without the forced additionalProperties: false. Remember that the layer wraps non-dict results as {"result": value}, so an output schema must describe the wrapper in that case, and that the schema is withheld from principals whose output pipeline is non-empty (3.7).

3.5 Annotation hints¶

ToolSpec._derive_hints (core.py lines 418-439) fills every hint left as None:

readOnlyHint is false if the capabilities include write_state, write_fs or exec, true otherwise.
destructiveHint is false.
openWorldHint is true when the capabilities include cloud_access or exec: such a tool reaches outside the loaded data.
idempotentHint is true for a read-only tool without cloud_access or exec, otherwise left unset. Two collections a minute apart can differ, so external tools are never idempotent by default.

Explicit decorator arguments win. What the derivation gives, from a run over several capability sets:

Capabilities	Wire annotations
`read_state`	readOnly true, destructive false, idempotent true, openWorld false
`read_state`, `write_state`	readOnly false, destructive false, openWorld false
`read_state`, `write_fs`	readOnly false, destructive false, openWorld false
`read_fs`	readOnly true, destructive false, idempotent true, openWorld false
`exec`	readOnly false, destructive false, openWorld true
`cloud_access`	readOnly true, destructive false, openWorld true
`reveal`	readOnly true, destructive false, idempotent true, openWorld false

Clients assume the worst for a missing hint (destructive, open world), which is why the layer fills them. The annotations also feed cloudg's own behavior: CachingMiddleware and RetryMiddleware only touch tools that are read-only and idempotent, so tools with cloud_access or exec stay out of both unless they declare idempotent=True, and --read-only drops tools annotated destructive.

3.6 URI templates¶

ResourceTemplateSpec accepts three variable forms:

Form	Matches	Use
`{var}`	one path segment (`[^/?]+`)	ids without slashes
`{+var}`	anything (`.+`)	RFC 6570 reserved expansion; ARNs and Azure resource ids
`{var*}`	anything	legacy spelling of the above, still accepted

match(uri) returns the variables URL-decoded; expand(**values) quotes {var} values completely and leaves /:@!$&'()*+,;= unescaped in {+var} values; variables lists the names. With the catalog's asset template:

spec = layer.registry.templates["cloudg://assets/{+ref}"]
spec.expand(ref="arn:aws:ec2:us-east-1:111111111111:instance/i-0bast")
# 'cloudg://assets/arn:aws:ec2:us-east-1:111111111111:instance/i-0bast'
spec.match(_)
# {'ref': 'arn:aws:ec2:us-east-1:111111111111:instance/i-0bast'}

and layer.read_resource() on that URI returned the bastion's detail.

3.7 `_meta` keys and errors¶

All layer-written _meta keys go through META_PREFIX (cloudg/) in core.py and layer.py: category, sensitivity, tags on definitions; transforms, duration_ms, error_code, error_data on results. middleware.py and cli.py spell the prefix out as a literal ("cloudg/transforms", "cloudg/cache", "cloudg/retries", "cloudg/category"); keep them in step if the prefix ever changes.

Error codes and their channels are tabulated in MCP.md section 13. The design rule is SEP-1303: input validation and handler failures are tool execution errors (isError), so the model can correct itself; only an unknown tool, resource or prompt, and malformed requests, are protocol errors.