Resources, prompts + sampling

MCP servers expose more than tools. MCPRegistry aggregates the full protocol surface across every configured server: resources (readable data the server publishes), prompts (server-authored message templates), and sampling (the server asking your client for an LLM completion).

from loomflow.mcp import MCPRegistry, MCPServerSpec
 
registry = MCPRegistry([
    MCPServerSpec.stdio("docs", "uvx", ["mcp-server-docs"]),
    MCPServerSpec.stdio("git",  "uvx", ["mcp-server-git", "--repo", "."]),
])
 
resources = await registry.list_resources()
readme    = await registry.read_resource("file:///project/README.md")
prompts   = await registry.list_prompts()
review    = await registry.get_prompt("code_review", {"language": "python"})

Resources

list_resources()

Aggregates resource listings across every server. Returns one dict per resource:

[{"uri": ..., "server": ..., "name": ..., "description": ..., "mime_type": ...}, ...]

uri is the server’s own URI when it’s unique across servers, or server:uri-qualified when two servers expose the same URI — the same bare-when-unique / qualified-always scheme used for tool names. Per-server failures are isolated: a flaky server is logged and skipped, never fatal.

read_resource(uri, *, server=None)

Reads one resource, routing to the owning server. Three addressing forms:

await registry.read_resource("file:///data/report.md")          # bare — unique owner
await registry.read_resource("docs:file:///data/report.md")     # server-qualified
await registry.read_resource("file:///tmpl/42.md", server="docs")  # explicit override

The explicit server= form also lets you read URIs the server never listed — resource-template expansions, for example. A bare URI owned by multiple servers raises MCPError telling you to qualify.

Return shape: text contents come back verbatim (a single block → str); blob contents become a {"mime": ..., "size": ...} placeholder; multiple content blocks come back as a list.

Prompts

list_prompts()

[{"name": ..., "server": ..., "description": ..., "arguments": ...}, ...]

Names use the same bare-when-unique / server.name-qualified scheme as tools, and get_prompt accepts either form.

get_prompt(name, arguments=None)

prompt = await registry.get_prompt("code_review", {"language": "python"})
# {"description": "...", "messages": [{"role": "user", "content": "..."}, ...]}

Text content is returned verbatim; binary blocks become minimal placeholders ([image: image/png, 34012 bytes]). Unknown names raise MCPError.

Sampling — servers calling your model

Some MCP servers request LLM completions from the client (sampling/createMessage) — e.g. a summarizing filesystem server that wants your model to condense a file. loomflow does not auto-wire a model here; sampling is opt-in and user-controlled via sampling_handler= on the server spec:

from loomflow import Agent
from loomflow.mcp import MCPRegistry, MCPServerSpec
from loomflow.model import AnthropicModel
 
sampler = AnthropicModel("claude-haiku-4-5")   # cheap model for server asks
 
async def handle_sampling(messages, model_preferences) -> str:
    text = "\n".join(str(getattr(m, "content", m)) for m in messages or [])
    reply, _tool_calls, _usage, _stop = await sampler.complete(
        [{"role": "user", "content": text}]
    )
    return reply
 
registry = MCPRegistry([
    MCPServerSpec.stdio(
        "summarizer", "uvx", ["mcp-server-summarizer"],
        sampling_handler=handle_sampling,
    ),
])

The handler is called as handler(messages, model_preferences) — messages is the SDK’s list of SamplingMessage objects, model_preferences the (possibly None) ModelPreferences. It may be sync or async and must return the completion text as str. Handler errors are returned to the server as JSON-RPC error data rather than crashing the session.

Where resources + prompts fit in an agent

Tools are wired into the loop automatically via Agent(tools=registry); resources and prompts are pull APIs you call from your own code — typically to seed context before a run:

style_guide = await registry.read_resource("file:///docs/style.md")
prompt = await registry.get_prompt("review_checklist")
 
agent = Agent(
    f"Follow this style guide:\n{style_guide}",
    model="claude-opus-4-7",
    tools=registry,
)

For registry lifecycle, fault isolation, and live tool-list refresh, see Server specs + registry.