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:
        answer = await client.query.run(
            query="What are the top whale movements on Base?",
            response_shape="answer_with_evidence",
        )
        print(answer.response)
        print(answer.summary)
        print(answer.evidence.facts if answer.evidence else None)

asyncio.run(main())

import asyncio
from ctxprotocol import ContextClient

async def main():
    async with ContextClient(api_key="sk_live_...") as client:
        tools = await client.discovery.search(
            "gas prices",
            mode="execute",
            surface="execute",
            require_execute_pricing=True,
        )
        session = await client.tools.start_session(max_spend_usd="1.00")
        method = tools[0].mcp_tools[0]
        result = await client.tools.execute(
            tool_id=tools[0].id,
            tool_name=method.name,
            args={"chainId": 1},
            session_id=session.session.session_id,
        )
        print(result.result)
        print(result.session)  # method_price, spent, remaining, max_spend, ...

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)

execute_tools = await client.discovery.search(
    "ethereum gas",
    mode="execute",
    surface="execute",
    require_execute_pricing=True,
)

featured = await client.discovery.get_featured(limit=5)
featured_execute = await client.discovery.get_featured(
    limit=5,
    mode="execute",
    require_execute_pricing=True,
)

session = await client.tools.start_session(max_spend_usd="2.50")

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",
    session_id=session.session.session_id,
)
print(result.method.execute_price_usd)  # explicit method price
print(result.session)  # method_price, spent, remaining, max_spend, ...

started = await client.tools.start_session(max_spend_usd="5.00")
print(started.session.session_id)
print(started.session.max_spend)

status = await client.tools.get_session("sess_123")
print(status.session.status)  # open | closed | expired
print(status.session.spent)

closed = await client.tools.close_session("sess_123")
print(closed.session.status)  # closed

answer = await client.query.run("What are the top whale movements on Base?")
print(answer.response)      # response text or summary
print(answer.tools_used)    # [QueryToolUsage(id, name, skill_calls)]
print(answer.cost)          # QueryCost(model_cost_usd, tool_cost_usd, total_cost_usd)
print(answer.orchestration_metrics)  # Optional first-pass / rediscovery metrics

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

print(answer.response_shape)  # "answer_with_evidence"
print(answer.summary)  # short machine-friendly summary
print(answer.evidence.facts if answer.evidence else None)
print(answer.artifacts.data_url if answer.artifacts else None)
print(answer.freshness.as_of if answer.freshness else None)
print(answer.confidence.level if answer.confidence else None)
print(answer.developer_trace.summary if answer.developer_trace else None)
print(
    answer.developer_trace.diagnostics.selection
    if answer.developer_trace and answer.developer_trace.diagnostics
    else None
)

result = await client.query.run(
    query="Which exchanges are seeing the largest BTC inflows and outflows over the last 24 hours?",
    response_shape="evidence_only",
)

print(result.response)  # machine-friendly summary for evidence_only
print(result.summary)
print(result.evidence.source_refs if result.evidence else None)
print(result.evidence.market_intelligence.venue_breakdown if result.evidence and result.evidence.market_intelligence else None)
print(result.stop_reason, result.issue_class, result.actions_taken)
print(result.usage.tools_used if result.usage else None)

async for event in client.query.stream(
    query="What are the top whale movements?",
    query_depth="fast",
):
    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,
    QueryOptions,
    QueryResult,
    QueryDeveloperTrace,
    QueryOrchestrationMetrics,
    QueryResponseEnvelopeEvidence,
    QueryResponseEnvelopeArtifacts,
    QueryResponseEnvelopeView,
    QueryResponseEnvelopeFreshness,
    QueryResponseEnvelopeConfidence,
    QueryResponseEnvelopeUsage,
    QueryClarificationPayload,
    QueryCapabilityMissPayload,
    QueryAssumptionMetadata,
    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  # listing-level response price metadata (legacy field name)
    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" (mode/eligibility/pricing/context)
    execute_eligible: bool | None
    execute_price_usd: str | None

TOOLS = [{
    "name": "get_price_history",
    "inputSchema": {
        "type": "object",
        "properties": {
            "symbol": {"type": "string", "default": "BTC", "examples": ["BTC", "ETH", "SOL"]},
            "interval": {"type": "string", "enum": ["1h", "4h", "1d"], "default": "1h", "examples": ["1h", "4h"]},
            "limit": {"type": "number", "default": 100, "examples": [50, 100, 200]},
        },
        "required": [],
    },
}]

class ExecutionResult(BaseModel):
    mode: Literal["execute"]
    result: Any
    tool: ToolInfo  # { id: str, name: str }
    method: ExecuteMethodInfo  # { name: str, execute_price_usd: str }
    session: ExecuteSessionSpend
    duration_ms: int

class ExecuteSessionSpend(BaseModel):
    mode: Literal["execute"] = "execute"
    session_id: str | None
    method_price: str
    spent: str
    remaining: str | None
    max_spend: str | None
    status: Literal["open", "closed", "expired"] | None
    expires_at: str | None
    close_requested: bool | None
    pending_accrued_count: int | None
    pending_accrued_usd: str | None

class QueryResult(BaseModel):
    response: str                        # prose answer or machine-friendly summary
    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)
    developer_trace: QueryDeveloperTrace | None  # Optional runtime trace + diagnostics
    orchestration_metrics: QueryOrchestrationMetrics | None  # Optional first-pass metrics
    response_shape: Literal["answer", "answer_with_evidence", "evidence_only"] | None
    summary: str | None
    evidence: QueryResponseEnvelopeEvidence | None
    artifacts: QueryResponseEnvelopeArtifacts | None
    view: QueryResponseEnvelopeView | None
    freshness: QueryResponseEnvelopeFreshness | None
    confidence: QueryResponseEnvelopeConfidence | None
    usage: QueryResponseEnvelopeUsage | None
    outcome_type: Literal["answer", "clarification_required", "capability_miss"]
    clarification: QueryClarificationPayload | None
    capability_miss: QueryCapabilityMissPayload | None
    assumption_made: QueryAssumptionMetadata | None

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.contrib.search import (
    build_contributor_search_validation_artifact,
    create_search_intent,
    extract_contributor_searches_from_developer_trace,
    merge_contributor_search_config,
    resolve_contributor_search,
)

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")