meshwork

AI interoperability middleware for multi-agent systems. meshwork provides typed channels, versioned shared context, capability-based routing, workflow orchestration, and MCP tool bridging — with pluggable transport and storage backends.

Install

pip install meshwork

Optional extras:

Extra	What it adds
meshwork[cli]	Command-line interface (click)
meshwork[dashboard]	Web dashboard (flask, flask-socketio)
meshwork[redis]	Redis transport and storage
meshwork[anthropic]	Anthropic Claude provider
meshwork[openai]	OpenAI provider
meshwork[all]	Everything above
meshwork[dev]	Testing tools (pytest)

Quick Start

from meshwork import Mesh, Workflow, configure_logging

configure_logging(level="INFO")

# Create a mesh with in-memory transport and SQLite storage
mesh = Mesh(transport="memory", storage="sqlite", db_path=":memory:")
mesh.start()

# Register agents with capabilities
mesh.register_agent(name="support", capabilities=["classify", "triage"])
mesh.register_agent(name="billing", capabilities=["refund", "invoice"])

# Send messages through channels
channel = mesh.create_channel("customer_issue")
channel.subscribe("billing")
channel.send(from_agent="support", content="Customer needs refund", recipient="billing")

msg = channel.receive("billing", timeout=5.0)
print(msg.content)  # "Customer needs refund"

# Share versioned context between agents
mesh.context.update("issue_123", {"status": "open", "priority": "high"}, updated_by="support")
print(mesh.context.get("issue_123"))  # {"status": "open", "priority": "high"}

# Define workflows with dependencies and conditions
wf = Workflow("handle_issue", mesh)

@wf.step(name="triage")
def triage(input_data, context):
    return {"context_updates": {"issue_type": "billing"}}

@wf.step(name="process", after="triage", condition=lambda ctx: ctx["issue_type"] == "billing")
def process(input_data, context):
    return {"context_updates": {"resolved": True}}

result = wf.run(input_data={"customer_id": "c_123"})
print(result["resolved"])  # True

mesh.stop()

Authentication

from meshwork import Mesh
from meshwork.auth import generate_token

mesh = Mesh(auth_secret="your-secret-key")
mesh.start()

token = generate_token("your-secret-key", "support", ["classify"])
mesh.register_agent(name="support", capabilities=["classify"], token=token)

MCP Bridge

from meshwork.bridge.mcp import MCPBridge

bridge = MCPBridge(mesh)
agent = bridge.connect("tools", server_command=["python", "mcp_server.py"])
# Tools discovered from the MCP server are registered as agent capabilities

Dashboard

export MESHWORK_SECRET_KEY="your-secret-key"
meshwork dashboard --port 7740

Or programmatically:

from meshwork.dashboard.app import create_app
app = create_app(mesh=mesh)

The dashboard exposes:

• GET /health — health check (status, agent/channel counts, storage reachability)
• GET /api/agents — registered agents
• GET /api/channels — active channels
• GET /api/context — context snapshot
• GET /api/messages/<channel> — message history
• WebSocket real-time updates (requires flask-socketio)

Logging

from meshwork import configure_logging

# Plain text (development)
configure_logging(level="DEBUG")

# Structured JSON (production)
configure_logging(level="INFO", json_format=True)

Architecture

Mesh (coordinator)
 |- Registry (capability-based agent discovery + routing)
 |- ContextBroker (versioned shared state with schema validation)
 |- Transport (in-memory | Unix socket | Redis pub/sub)
 |- Storage (SQLite | Redis)
 |- Channels (typed message delivery with ack/dead-letter)
 |- Workflow (decorator-based, parallel branches, failure recovery)
 |- MCPBridge (MCP tool server integration)

Contributing

See AGENTS.md for development guidelines.

License

MIT