Python SDK Reference

pip install ctxprotocol[fastapi]

import asyncio
from ctxprotocol import ContextClient

async def main():
    async with ContextClient(api_key="sk_live_...") as client:
        # Pay-per-response: Ask a question, get a curated answer
        answer = await client.query.run("What are the top whale movements on Base?")
        print(answer.response)

        # Pay-per-request: Execute a specific tool for raw data
        tools = await client.discovery.search("gas prices")
        result = await client.tools.execute(
            tool_id=tools[0].id,
            tool_name=tools[0].mcp_tools[0].name,
            args={"chainId": 1},
        )
        print(result.result)

asyncio.run(main())

import os
from ctxprotocol import ContextClient

# Production
client = ContextClient(api_key=os.environ["CONTEXT_API_KEY"])

# Local development
client = ContextClient(
    api_key="sk_test_...",
    base_url="http://localhost:3000",
)

tools = await client.discovery.search("ethereum gas", limit=10)

featured = await client.discovery.get_featured(limit=5)

result = await client.tools.execute(
    tool_id="uuid-of-tool",
    tool_name="get_gas_prices",
    args={"chainId": 1},
    idempotency_key="2bb4bdcb-8609-43f6-af13-75279186de70",
)

answer = await client.query.run("What are the top whale movements on Base?")
print(answer.response)      # AI-synthesized text
print(answer.tools_used)    # [QueryToolUsage(id, name, skill_calls)]
print(answer.cost)          # QueryCost(model_cost_usd, tool_cost_usd, total_cost_usd)

answer = await client.query.run(
    query="Analyze whale activity on Base",
    model_id="glm-model",
    include_data=True,
    include_data_url=True,
    idempotency_key="6e7f1389-f72f-41d9-bf26-0608a4d8be87",
)

async for event in client.query.stream("What are the top whale movements?"):
    if event.type == "tool-status":
        print(f"Tool {event.tool.name}: {event.status}")
    elif event.type == "text-delta":
        print(event.delta, end="")
    elif event.type == "done":
        print(f"\nTotal cost: {event.result.cost.total_cost_usd}")

from ctxprotocol import (
    # Auth utilities for tool contributors
    verify_context_request,
    is_protected_mcp_method,
    is_open_mcp_method,

    # Client types
    ContextClientOptions,
    Tool,
    McpTool,
    ExecuteOptions,
    ExecutionResult,
    ContextErrorCode,

    # Auth types (for MCP server contributors)
    VerifyRequestOptions,

    # Context types (for MCP server contributors receiving injected data)
    ContextRequirementType,
    HyperliquidContext,
    PolymarketContext,
    WalletContext,
    UserContext,
)

class Tool(BaseModel):
    id: str
    name: str
    description: str
    price: str
    category: str | None
    is_verified: bool | None
    mcp_tools: list[McpTool] | None

class McpTool(BaseModel):
    name: str
    description: str
    input_schema: dict[str, Any] | None  # JSON Schema for arguments
    output_schema: dict[str, Any] | None  # JSON Schema for response
    meta: dict[str, Any] | None  # alias: "_meta" (context + pacing metadata)

class ExecutionResult(BaseModel):
    result: Any
    tool: ToolInfo  # { id: str, name: str }
    duration_ms: int

class QueryResult(BaseModel):
    response: str                        # AI-synthesized answer
    tools_used: list[QueryToolUsage]     # [{ id, name, skill_calls }]
    cost: QueryCost                      # { model_cost_usd, tool_cost_usd, total_cost_usd }
    duration_ms: int
    data: Any | None                     # Optional execution data (include_data=True)
    data_url: str | None                 # Optional blob URL (include_data_url=True)

from ctxprotocol import CONTEXT_REQUIREMENTS_KEY

# Context types supported by the marketplace
ContextRequirementType = Literal["polymarket", "hyperliquid", "wallet"]

# Usage: Declare context requirements in _meta at the tool level
TOOLS = [{
    "name": "analyze_my_positions",
    "description": "Analyze your positions with personalized insights",
    "_meta": {
        "contextRequirements": ["hyperliquid"],
        "rateLimit": {
            "maxRequestsPerMinute": 30,
            "cooldownMs": 2000,
            "maxConcurrency": 1,
            "supportsBulk": True,
            "recommendedBatchTools": ["get_portfolio_snapshot"],
            "notes": "Hobby tier: prefer snapshot endpoints over loops.",
        },
    },
    "inputSchema": {
        "type": "object",
        "properties": {
            "portfolio": {
                "type": "object",
                "description": "Portfolio context (injected by platform)",
            },
        },
        "required": ["portfolio"],
    },
}]

class HyperliquidContext(BaseModel):
    wallet_address: str
    perp_positions: list[HyperliquidPerpPosition]
    spot_balances: list[HyperliquidSpotBalance]
    open_orders: list[HyperliquidOrder]
    account_summary: HyperliquidAccountSummary
    fetched_at: str

class PolymarketContext(BaseModel):
    wallet_address: str
    positions: list[PolymarketPosition]
    open_orders: list[PolymarketOrder]
    total_value: float | None
    fetched_at: str

class WalletContext(BaseModel):
    address: str
    chain_id: int
    native_balance: str | None

from ctxprotocol import ContextClient, ContextError

try:
    result = await client.tools.execute(...)
except ContextError as e:
    match e.code:
        case "no_wallet":
            # User needs to set up wallet
            print(f"Setup required: {e.help_url}")
        case "insufficient_allowance":
            # User needs to set a spending cap
            print(f"Set spending cap: {e.help_url}")
        case "payment_failed":
            # Insufficient USDC balance
            pass
        case "execution_failed":
            # Tool execution error
            pass

from fastmcp import FastMCP
from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.server.dependencies import get_http_headers
from fastmcp.exceptions import ToolError
from ctxprotocol import verify_context_request, ContextError

mcp = FastMCP("my-tool")

class ContextProtocolAuth(Middleware):
    """Verify Context Protocol JWT on tool calls only."""
    
    async def on_call_tool(self, context: MiddlewareContext, call_next):
        headers = get_http_headers()
        try:
            await verify_context_request(
                authorization_header=headers.get("authorization", "")
            )
        except ContextError as e:
            raise ToolError(f"Unauthorized: {e.message}")
        return await call_next(context)

mcp.add_middleware(ContextProtocolAuth())

@mcp.tool
def get_data(query: str) -> dict:
    return {"result": "..."}

if __name__ == "__main__":
    mcp.run(transport="http", port=3000)

from fastapi import FastAPI, Request, Depends, HTTPException
from ctxprotocol import create_context_middleware, ContextError

app = FastAPI()
verify_context = create_context_middleware(audience="https://your-tool.com/mcp")

@app.post("/mcp")
async def handle_mcp(request: Request, context: dict = Depends(verify_context)):
    # context contains verified JWT payload (on protected methods)
    # None for open methods like tools/list
    body = await request.json()
    # Handle MCP request...

from ctxprotocol import verify_context_request, is_protected_mcp_method, ContextError

# Check if a method requires auth
if is_protected_mcp_method(body["method"]):
    try:
        payload = await verify_context_request(
            authorization_header=request.headers.get("authorization"),
            audience="https://your-tool.com/mcp",  # optional
        )
        # payload contains verified JWT claims
    except ContextError:
        raise HTTPException(status_code=401, detail="Unauthorized")