physarum-sdk 0.2.0

Physarum Intelligence Network Python SDK — MCP tool routing, model selection, and cost optimization

physarum-sdk

Python SDK for the Physarum Intelligence Network — real-time MCP tool routing, model selection, and cost optimization powered by network-wide telemetry and Physarum-inspired conductivity algorithms.

What it does

• Routes tool calls to the best-performing implementation based on live success rates, latency, and quality signals collected across all tenants
• Tracks every tool execution with zero-overhead telemetry batching
• Works with any Python AI framework — LangChain, LlamaIndex, raw OpenAI/Anthropic calls
• Fails open — if the API is unavailable, falls back to your configured static priorities

Installation

pip install physarum-sdk

Quick start

from physarum import PhysarumClient, PhysarumConfig

client = PhysarumClient(PhysarumConfig(
    api_key=os.environ["PHYSARUM_API_KEY"],
    tenant_id=os.environ["PHYSARUM_TENANT_ID"],
    ingestion_base_url="https://api.physarum.network",
    recommendation_base_url="https://api.physarum.network",
    mode="SHADOW",  # Start with SHADOW, graduate to CONTROLLED
))

# Ask Physarum which tool to use
from physarum.types import RouteRequest

decision = client.select_tool(RouteRequest(
    task_category="payment_flow",
    candidate_tools=["stripe", "paypal", "razorpay"],
    action_class="SIDE_EFFECT_CRITICAL",
))

print(decision["selected_tool"])  # e.g. "stripe"
print(decision["reason"])         # "controlled_mode" | "shadow_mode" | ...

client.shutdown()

Operating modes

Mode	Behaviour
SHADOW	Observes only. Records telemetry but never changes which tool is called. Zero risk, full learning.
ADVISORY	Calls the recommendation API and logs the suggestion but still runs your default tool.
CONTROLLED	Physarum selects the tool. The network's best recommendation is used for every call.

Start with SHADOW to accumulate signal, then graduate to CONTROLLED once you trust the data.

Manual tool wrapping

Wrap any function call to automatically record success, latency, and error telemetry:

from physarum.types import WrapToolCallInput

outcome = client.wrap_tool_call(WrapToolCallInput(
    tool_id="stripe",
    tool_name="stripe",
    task_category="payment_flow",
    action_class="SIDE_EFFECT_CRITICAL",
    session_id_hash="hashed-session-id",
    execute=lambda: stripe.charge(amount=9900, currency="usd"),
))

print(outcome.result)       # whatever stripe.charge returned
print(outcome.telemetry)    # full telemetry dict, already flushed to Physarum

LangChain integration

from langchain.tools import StructuredTool
from physarum.types import WrapToolCallInput

def make_physarum_tool(client, name, description, func, task_category, action_class):
    def wrapped(**kwargs):
        outcome = client.wrap_tool_call(WrapToolCallInput(
            tool_id=name,
            tool_name=name,
            task_category=task_category,
            action_class=action_class,
            session_id_hash="your-session-hash",
            execute=lambda: func(**kwargs),
        ))
        return outcome.result

    return StructuredTool.from_function(
        func=wrapped,
        name=name,
        description=description,
    )

search_tool = make_physarum_tool(
    client,
    name="search_products",
    description="Search the product catalogue",
    func=search_products_api,
    task_category="product_search",
    action_class="READ_ONLY",
)

Context enrichment

Pass context to improve routing accuracy. Physarum learns per-country, per-domain, and per-locale performance:

from physarum.types import RouteRequest, ContextInput

decision = client.select_tool(RouteRequest(
    task_category="payment_flow",
    candidate_tools=["stripe", "razorpay"],
    action_class="SIDE_EFFECT_CRITICAL",
    context=ContextInput(
        country_code="IN",       # India — Razorpay likely performs better
        domain="e-commerce",
        locale="en-IN",
        model_id="claude-sonnet-4-6",
        time_of_day_utc="14:30",
    ),
))

Model routing

from physarum.types import ModelRouteRequest

result = client.get_model_routes(ModelRouteRequest(
    task_category="code_debug",
    candidate_models=["claude-opus-4-6", "claude-sonnet-4-6", "gpt-4o"],
))

best_model = result.recommendations[0]["model_id"]

Cost optimization

from physarum.types import CostOptimizeRequest

result = client.get_cost_optimized_path(CostOptimizeRequest(
    task_category="document_summarization",
    candidate_tools=["gpt-4o", "claude-sonnet-4-6", "gemini-flash"],
    quality_floor=0.8,     # minimum acceptable quality score
    budget_tokens=50_000,  # max tokens to spend
))

MCP server discovery

# Get all MCP servers registered on the network
servers = client.get_mcp_servers()

# Filter by task category
payment_servers = client.get_mcp_servers(task_category="payment_flow")

Static fallback priorities

Configure a deterministic fallback order used when the recommendation API is unavailable:

client = PhysarumClient(PhysarumConfig(
    # ...
    local_static_priorities=["stripe", "paypal", "razorpay"],
    local_static_priorities_by_task_category={
        "payment_flow_india": ["razorpay", "stripe"],
    },
))

Configuration reference

from physarum import PhysarumConfig

config = PhysarumConfig(
    api_key="...",
    tenant_id="...",
    ingestion_base_url="https://api.physarum.network",
    recommendation_base_url="https://api.physarum.network",
    mode="SHADOW",                          # "SHADOW" | "ADVISORY" | "CONTROLLED"
    request_timeout_ms=5000,                # default: 5000
    telemetry_batch_size=50,                # events per flush, default: 50
    telemetry_flush_interval_ms=2000,       # default: 2000ms
    local_static_priorities=[],             # fallback tool order
    local_static_priorities_by_task_category={},
)

Action classes

Value	Use when
READ_ONLY	Tool only reads data — search, lookup, fetch
IDEMPOTENT_WRITE	Safe to retry — upsert, idempotent create
SIDE_EFFECT_CRITICAL	Must not be retried blindly — payment, email send, webhook

Shutdown

Always call shutdown() before your process exits to flush buffered telemetry:

import atexit
atexit.register(client.shutdown)

Or use as a context manager pattern:

try:
    result = client.wrap_tool_call(...)
finally:
    client.shutdown()

License

MIT