Skip to content

Framework-free export

Section 9 · From docs/MCP.md on GitHub

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.