state-mesh 0.1.1

Async-native agentic pipeline runtime for Python backend engineers

State-Mesh

Async-native agentic pipeline runtime for Python backend engineers.

Why State-Mesh

• Typed state — Pydantic models flow through every step, no silent dict mutations
• Built-in MCP — register MCP servers once, every step can call any tool
• Production-ready — guardrails, structured output retry, OTel tracing, pipeline resumption out of the box

Features

• Async-first step execution with retry, timeout, and branching
• Parallel step execution with asyncio.gather
• Structured output contracts with automatic LLM retry
• Guard chain — schema, PII, confidence, toxicity guards built in
• MCP tool bus (SSE + stdio transport)
• OpenTelemetry tracing with swappable exporters
• Redis state backend with pipeline resumption
• FastAPI integration — mount_pipeline, stream_pipeline, pipeline_router
• Human-in-the-loop with timeout and fallback

Install

pip install state-mesh
pip install "state-mesh[redis]"   # Redis backend
pip install "state-mesh[otel]"    # OpenTelemetry

Quickstart

from state_mesh import step, Pipeline, Context, OutputContract
from pydantic import BaseModel

class Result(BaseModel):
    summary: str

@step(output_contract=OutputContract(target_schema=Result))
async def summarise(prompt: str) -> str:
    return await my_llm(prompt)

pipeline = Pipeline(steps=[summarise], name="summarise")
result = await pipeline.run(Context(state={}))

License

MIT