physarum-sdk 0.2.4
Physarum Intelligence Network Python SDK — MCP tool routing, model selection, and cost optimization
License: MIT
Requires Python: >=3.10
Keywords: mcp, ai, routing, tool-selection, intelligence

# 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.

[![PyPI version](https://img.shields.io/pypi/v/physarum-sdk)](https://pypi.org/project/physarum-sdk/)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)
[![Python](https://img.shields.io/pypi/pyversions/physarum-sdk)](https://pypi.org/project/physarum-sdk/)

## 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

```bash
pip install physarum-sdk
```

## Quick start

```python
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:

```python
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

```python
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:

```python
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

```python
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

```python
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

```python
# 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:

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

## Configuration reference

```python
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:

```python
import atexit
atexit.register(client.shutdown)
```

Or use as a context manager pattern:

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

## License

MIT