Skip to content

Python SDK

MIT licensed. Runtime dependency: httpx. The official mcp package and pydantic-ai are optional extras.

Terminal window
uv add "protogrid[mcp]" # + [pydantic-ai] for the toolset
from protogrid import ProtogridClient, AsyncProtogridClient
registry = ProtogridClient("https://api.protogrid.dev") # api_key= (default: PROTOGRID_API_KEY), timeout= optional
hits = registry.search("send an email", class_=["R0", "R1"], min_trust=60, exclude_flags=["multi-version-spam"])
descriptor = registry.get_server(hits["results"][0]["name"])
tools = registry.list_all_tools(hits["results"][0]["name"])
conn = registry.get_connection(hits["results"][0]["name"]) # mcpServers block with ${SECRET} placeholders
cli = registry.get_connection(hits["results"][0]["name"], "claude-code-cli")

Responses are plain dicts typed as TypedDicts. Errors are ProtogridError with status, code and retry_after. AsyncProtogridClient has the same methods, awaitable.

check(url) probes any remote MCP server URL once, listed or not, and returns its quality checks and its readiness for the Claude and OpenAI directories (see Check your server). It waits up to 90 seconds by default (wait=); a check still running then can be read later with get_check(id).

c = registry.check("https://mcp.example.com/mcp")
if c["status"] == "done":
for d in c["result"]["readiness"]:
print(d["name"], d["summary"]["blockers"], "blockers")
print("quality", c["result"]["quality"]["score"], c["page"])

A refused URL raises invalid_url; an exhausted hourly allowance raises check_quota_exceeded with retry_after.

from protogrid import placeholders_in, substitute_secrets, secrets_satisfied
placeholders_in(conn["connection"]) # ["AGENTDM_TOKEN"]
secrets_satisfied(conn, os.environ) # False until AGENTDM_TOKEN is set
substitute_secrets(conn["connection"], os.environ) # deep copy; raises MissingSecretsError
import os
from protogrid import ProtogridClient, find_connectable, open_session
registry = ProtogridClient("https://api.protogrid.dev")
found = find_connectable(registry, "get the weather for a city", secrets=os.environ)
async with open_session(found.connection, os.environ) as session:
tools = (await session.list_tools()).tools
out = await session.call_tool("get_weather", {"city": "Madrid"})

open_session yields an initialized mcp.ClientSession over streamable HTTP, SSE or stdio. find_connectable returns the first hit that is R0, R1 with secrets present, or R2 with stored tokens (token_store=); local packages only with allow_local=True. afind_connectable is the async twin.

from protogrid import FileTokenStore, OAuthOptions, loopback_consent, open_session
store = FileTokenStore(".mcp-tokens.json")
consent = loopback_consent(port=8765) # prints the URL a human opens once
conn = registry.get_connection("ac.snag/snag")
async with open_session(conn, oauth=OAuthOptions(store=store, consent=consent, client_name="my agent")) as session:
...

The official mcp package runs discovery, dynamic registration (or a Client ID Metadata Document via client_metadata_url), PKCE and refresh inside its OAuthClientProvider; the SDK supplies the provider from your TokenStore (get/set/delete; MemoryTokenStore, FileTokenStore included) and the consent handler. Headless agents use manual_consent(redirect_url, on_authorization_url, wait_for_code). oauth_provider(...) returns the httpx.Auth for hand-built transports.

from pydantic_ai import Agent
from protogrid import ProtogridClient, find_connectable, to_pydantic_ai
found = find_connectable(ProtogridClient("https://api.protogrid.dev"), "get the weather for a city", secrets=os.environ)
agent = Agent("anthropic:claude-sonnet-5", toolsets=[to_pydantic_ai(found.connection, os.environ)])
async with agent:
result = await agent.run("What is the weather in Madrid right now?")

to_pydantic_ai returns an MCPToolset over the matching FastMCP transport; pass auth=oauth_provider(...) for R2 servers. The repository’s examples/pydantic_ai_agent.py runs the same with PydanticAI’s TestModel, so it needs no LLM key.

Terminal window
cd sdks/python && uv sync
uv run python examples/find_and_call.py "get the current weather for a city" Madrid
uv run python examples/pydantic_ai_agent.py "get the current weather for a city"
# both use the public registry; prefix with REGISTRY_URL=http://localhost:8080 for a local stack