Framework-free export¶
For gateways, agent frameworks or custom servers that do not use an MCP SDK,
export_definitions returns the wire definitions as one principal sees them, plus async
handlers that still go through the layer.
from cloudg.mcp import export_definitions # or cloudg.mcp.adapters.export_definitions
defs = export_definitions(layer, principal=None) # None = Principal.local()
ExportedDefinitions fields:
| Field | Content |
|---|---|
server |
{"name", "version", "instructions"} |
tools, resources, resource_templates, prompts |
MCP wire dicts, snapshot for the given principal |
handlers |
{method: async fn(params, *, principal=None) -> result dict} for tools/list, tools/call, resources/list, resources/templates/list, resources/read, prompts/list, prompts/get, completion/complete |
tool_functions |
{exposed tool name: async fn(**arguments) -> CallToolResult dict}, bound to the export's principal |
await defs.dispatch(method, params, principal=...) runs one handler. An unknown method raises
KeyError; protocol-level failures (unknown tool, unknown resource, prompt argument errors)
raise MCPLayerError subclasses; defs.as_dict() returns the definitions for serializing.
From a run against a small test registry with prefix="cloudg_":
print(sorted(defs.handlers))
# ['completion/complete', 'prompts/get', 'prompts/list', 'resources/list', 'resources/read',
# 'resources/templates/list', 'tools/call', 'tools/list']
print(sorted(defs.tool_functions))
# ['cloudg_boom', 'cloudg_custom_result', 'cloudg_progress_async', 'cloudg_reset', 'cloudg_slow_sync']
await defs.dispatch("tools/call", {"name": "cloudg_custom_result", "arguments": {}})
# {'content': [{'type': 'text', 'text': 'two assets'},
# {'type': 'resource_link', 'uri': 'cloudg://workspace', 'name': 'workspace',
# 'mimeType': 'application/json', 'annotations': {'priority': 0.5}}],
# 'isError': False, 'structuredContent': {'count': 2}, '_meta': {'cloudg/duration_ms': 0}}
await defs.tool_functions["cloudg_progress_async"](steps=1)
# {'content': [...], 'isError': False, 'structuredContent': {'result': [1, 2, 3]}, ...}
Function-calling formats for model APIs, each fn(layer, principal=None) -> list:
from cloudg.mcp.adapters import to_anthropic_tools, to_langchain_tools, to_openai_tools
to_openai_tools(layer)[0]
# {'type': 'function',
# 'function': {'name': 'cloudg_slow_sync', 'description': 'Sleeps in a worker thread.',
# 'parameters': {'additionalProperties': False,
# 'properties': {'seconds': {'default': 1.0, 'description': 'How long',
# 'maximum': 10, 'minimum': 0,
# 'type': 'number'}},
# 'type': 'object'}}}
to_anthropic_tools(layer)[0]
# {'name': 'cloudg_slow_sync', 'description': 'Sleeps in a worker thread.',
# 'input_schema': {...same schema...}}
When the model asks for a tool, execute it with await layer.call_tool(name, arguments,
principal=...) and send back the text content, or the structured content if your API
accepts JSON. A loop with the Anthropic SDK:
import json
import anthropic
client = anthropic.AsyncAnthropic()
tools = to_anthropic_tools(layer, principal)
messages = [{"role": "user", "content": "Which assets are internet exposed?"}]
while True:
reply = await client.messages.create(model=MODEL, max_tokens=2048, tools=tools,
messages=messages)
messages.append({"role": "assistant", "content": reply.content})
calls = [b for b in reply.content if b.type == "tool_use"]
if not calls:
break
results = []
for call in calls:
res = await layer.call_tool(call.name, call.input, principal=principal)
results.append({"type": "tool_result", "tool_use_id": call.id, "is_error": res.is_error,
"content": res.content[0].text if res.content else ""})
messages.append({"role": "user", "content": results})
MODEL is whichever model id you use. I did not run this loop for the guide; the
to_anthropic_tools output above is real.
to_langchain_tools(layer, principal=None) returns LangChain StructuredTool objects with
response_format="content_and_artifact" (text as content, structured result as artifact).
It needs pip install langchain-core, which was not installed in the test environment, so
this function is untested here.