⚡ Moven Python SDK (moven-sdk)

The Synchronous Circuit Breaker for Autonomous AI Agents in Python. Real-time, in-process safety fuses that detect runaway tool loops, hallucinated parameters, and cost spikes before your credit card burns.

💡 Why Moven?

Observability platforms (LangSmith, Langfuse, Helicone) record what happened after your agent finishes. If your agent enters an unhandled 150-step loop at 2 AM, traditional tools show you a $300 bill in the morning.

Moven sits synchronously in the execution loop. It evaluates deterministic heuristics in < 0.8ms on every tool call and trips the fuse mid-flight before money burns.

✨ Core Capabilities

• ⚡ Zero-Latency In-Memory Hot-Path: In-memory static heuristics evaluate in < 0.8ms without network proxies.
• 🔁 Canonical Deep Parameter Hashing: SHA-256 canonical serialization detects duplicate parameter loops regardless of dict key order.
• 💸 Dynamic Live Pricing Engine: Real-time token math synced from https://api.moven.dev/v1/models calculates exact dollar savings when loops are intercepted.
• 🛡️ Zero-Trust Hallucination Guard: Intercepts unpopulated placeholder arguments (TODO_..., REPLACE_ME, None) and non-existent schema parameters.
• ⏪ Ctrl+Z Step Checkpoints: Automatically snapshots agent state & prompts before every tool execution for instant time-travel rewinds.
• 🤖 Multi-Model Dynamic Auto-Fallback: Automatically tracks token burn rates and provides in-memory model tiering.
• 🌐 Python Framework Adapters: Native support for LangChain, LangGraph, CrewAI, AutoGen, OpenAI Python SDK, Anthropic Python SDK, and custom functions via @breaker.protect.

📦 Installation

pip install moven-sdk

🚀 Quick Start Examples

1. Protect Agent Tools with Decorator

from moven_sdk import MovenCircuitBreaker, BreakerConfig

# Initialize in-memory circuit breaker
breaker = MovenCircuitBreaker(
    BreakerConfig(
        project_id="a263283f-2d0b-4ce1-a40d-37103f09a160",
        max_repeats=3,               # Trip fuse after 3 identical tool calls
        spend_ceiling_usd=2.00,       # Stop runaway spend at $2.00
        max_turns=15,                 # Max recursion depth
        model_name="openai/gpt-4o",   # Used for real-time dollar calculation
    )
)

# Decorate any tool function
@breaker.protect
def query_vector_db(query: str, top_k: int = 5):
    # Your agent's tool execution logic
    return vector_store.similarity_search(query, k=top_k)

2. LangChain & LangGraph Callback Integration

from moven_sdk import MovenCircuitBreaker, BreakerConfig
from moven_sdk.adapters.langchain import MovenLangChainCallbackHandler
from langchain_openai import ChatOpenAI

# Initialize circuit breaker and callback handler
breaker = MovenCircuitBreaker(
    BreakerConfig(
        project_id="my-langgraph-project",
        max_repeats=3,
        spend_ceiling_usd=1.50,
    )
)
handler = MovenLangChainCallbackHandler(breaker)

# Attach callback handler to your LLM or LangGraph workflow
llm = ChatOpenAI(model="gpt-4o", callbacks=[handler])

3. CrewAI Tool Integration

from crewai import Agent, Task, Crew
from crewai.tools import tool
from moven_sdk import MovenCircuitBreaker, BreakerConfig

breaker = MovenCircuitBreaker(BreakerConfig(project_id="my-crewai-fleet"))

@tool("Search Internet")
@breaker.protect
def search_internet(query: str) -> str:
    """Searches the internet for relevant news."""
    return search_api.run(query)

4. Dynamic Live Model Pricing & Dollar Savings

Moven syncs live rates directly from https://api.moven.dev/v1/models and calculates accurate token and dollar savings when an infinite loop is aborted:

from moven_sdk import MovenDynamicPricingEngine

# 0ms in-memory lookup synced with OpenRouter catalog
rates = MovenDynamicPricingEngine.get_model_rates("anthropic/claude-3.5-sonnet")
print(f"Claude Sonnet Input Rate: ${rates['prompt']}/1M tokens")

# Exact dollar savings calculation on tripped loops
savings = MovenDynamicPricingEngine.calculate_money_saved(
    model_name="openai/gpt-4o",
    total_tool_calls_made=5
)

print(f"Prevented Spend: ${savings['money_saved']} USD ({savings['prevented_tokens']:,} tokens prevented)")

🛠️ Python Framework Adapters Reference

⚙️ Configuration Reference (BreakerConfig)

from moven_sdk import BreakerConfig

config = BreakerConfig(
    project_id="default",                       # Moven project ID
    agent_name="production_agent",             # Agent identifier
    max_repeats=3,                              # Trip on N consecutive identical tool invocations
    spend_ceiling_usd=2.00,                     # Maximum hard dollar spend ceiling
    max_turns=50,                               # Maximum execution steps per session
    model_name="openai/gpt-4o",                 # Model used for pricing calculation
    endpoint="https://api.moven.dev/events",    # Telemetry streaming endpoint
    strict_placeholders=True,                   # Block unpopulated template args (TODO_...)
    auto_heal_github=False,                     # Dispatch automated AST GitHub pull request
)

🧪 Running Tests

python -m pytest tests/

📜 License

MIT © Moven AI