agenthinkmesh/agenthink-mesh

🕸️ AgenThink Mesh

The Universal Positioning System (UPS) for AI. The open protocol connecting every AI agent on earth.

  ◆ Research Agent ──┐
  ◆ Legal Agent    ──┼──► [ MESH ORCHESTRATOR ] ──► Mission Output
  ◆ Strategy Agent ──┘         │
                          State-Tree
                          (The Vault)

---

What is AgenThink Mesh?

Just as GPS gives every car a shared coordinate system — AgenThink Mesh gives every AI agent a shared mission state.

Mesh is an open-source service mesh and orchestration protocol for multi-agent AI systems. It solves the three hardest problems in production agentic architectures:

Problem	Mesh Solution
"Which agent should handle this?"	Universal Agent Directory — discovery by capability, trust score & latency
"What did the previous agent know?"	State-Relay Protocol — JSON State-Packets travel with the mission
"What if an agent fails or hallucinates?"	Hot-Swap Engine — automatic recalculation, zero mission interruption

---

What's New in v0.1.0

1. @mesh_node Decorator

Turn any Python function into a fully Mesh-compatible agent in 4 lines:

from sdk.python.mesh_node import mesh_node

@mesh_node(
    name="Research Agent",
    nan_address="nan://global/research/my-agent/1.0",
    domain="research",
    skills=["web_search", "synthesis"],
)
async def research(context: dict) -> str:
    return f"Findings: {context.get('objective')}"

The decorator automatically:

• Wraps input/output into State-Packet JSON format
• Attaches trace_id, schema_version, confidence_score, timestamp
• Sanitizes kwargs — strips internal mesh metadata before passing to your function
• Persists result to The Vault
• Emits real-time telemetry events

---

2. Hot-Swap Recalculation Engine

When an agent fails, returns low confidence, or breaches its SLA — Mesh recalculates mid-mission and hot-swaps to the best available replacement. The mission never stops.

Agent A fails
     ↓
HotSwapOrchestrator.evaluate()
     ↓
Query registry → score candidates by trust tier + health score
     ↓
Hot-swap to Agent B with SAME State-Packet (full context preserved)
     ↓
Mission continues — zero interruption

Reroute triggers:

• Agent returns error or times out
• Confidence score below configurable threshold (default 0.70)
• Grounding Gate blocks hallucinated output

Infinite-loop prevention: agents with health score below 0.3 are permanently excluded from reroute candidates.

---

3. The Vault — Session Persistence

Every State-Packet is stored by mission_id. Any agent joining mid-mission instantly reads the full history — no re-briefing, no context loss.

# Agent reads the full mission map on join
prior_packets = await vault.get_mission_packets(mission_id)

• SQLite (default) — zero config, WAL mode enabled, works everywhere
• Redis (production) — set MESH_VAULT_BACKEND=redis
• Parameterized queries throughout — no SQL injection risk
• WAL mode + busy_timeout=5000 — safe for concurrent agent writes

---

4. Live Telemetry Stream — /telemetry

Server-Sent Events stream of all agent activity. Connect any dashboard:

# Stream all events
curl -N http://localhost:8000/telemetry

# Filter by mission
curl -N "http://localhost:8000/telemetry?mission_id=msn_abc123"

Events streamed: agent_start, agent_complete, grounding_check, reroute_triggered, mission_complete, health_ping

---

Repository Structure

agentthink-mesh/
│
├── core/
│   ├── main.py                    # App entrypoint + all API routes
│   ├── models.py                  # StatePacket, MissionStateTree, AgentManifest
│   ├── relay.py                   # NAN-RELAY/1.0 — State-Packet handoff engine
│   ├── rerouter.py                # Dynamic rerouting + trust score updates
│   ├── hotswap.py                 # ★ NEW — Hot-swap recalculation engine
│   └── middleware/
│       └── grounding_gate.py      # Hallucination firewall (MCP web search)
│
├── registry/
│   ├── agents.yaml                # Agent manifest — ID, roles, trust scores
│   └── registry.py                # Registry loader + domain/skill queries
│
├── sdk/
│   ├── python/
│   │   ├── mesh_node.py           # ★ NEW — @mesh_node decorator
│   │   ├── vault.py               # ★ NEW — The Vault (SQLite + Redis)
│   │   ├── telemetry.py           # ★ NEW — SSE /telemetry stream
│   │   └── mesh_sdk.py            # MeshAgent + MeshClient classes
│   └── typescript/
│       └── mesh-sdk.ts            # TypeScript SDK
│
├── ui/
│   └── dashboard.py               # Rich terminal Mission Control dashboard
│
├── tests/
│   └── highway_patrol/
│       └── test_grounding.py      # CI schema + latency + relay tests
│
├── docs/
│   └── protocol_spec.md           # NAN-RELAY/1.0 full specification
│
├── market_entry_demo.py           # 3-agent chained mission demo
├── requirements.txt
└── .gitlab-ci.yml                 # Highway Patrol CI/CD pipeline

---

Quick Start

# 1. Clone
git clone https://gitlab.com/agenthink/mesh.git
cd mesh

# 2. Install
pip install -r requirements.txt

# 3. Start the Orchestrator
uvicorn core.main:app --reload --port 8000

# 4. Open API docs
# http://localhost:8000/mesh/docs

# 5. Stream telemetry (new terminal)
curl -N http://localhost:8000/telemetry

# 6. Run the 3-agent market entry demo (new terminal)
python market_entry_demo.py

---

State-Packet Schema (v1.0)

{
  "packet_id":      "sp_8f3a92c1d",
  "schema_version": "1.0",
  "trace_id":       "trc_4a7f2b9c1e3d5a8f",
  "mission_id":     "msn_001",
  "sequence":       2,
  "from_agent":     "nan://global/research/web-researcher/2.1",
  "to_agent":       "nan://kw/legal/contract-analyst/1.5",
  "context": {
    "Research Agent": { "output": "...", "confidence": 0.91 },
    "_summary": "Rolling mission context..."
  },
  "confidence":     0.91,
  "grounding":      { "result": "passed", "sources_verified": 3 },
  "issued_by":      "nan://mesh/orchestrator/core/1.0"
}

---

API Endpoints

Method	Endpoint	Description
GET	/mesh/health	Orchestrator health + vault status
GET	/mesh/agents	List registry agents (?domain= filter)
POST	/mesh/missions	Dispatch a multi-agent mission
GET	/mesh/missions/{id}	Mission state + recent packets
GET	/mesh/vault/{mission_id}	All State-Packets from Vault
GET	/telemetry	SSE — live agent event stream
GET	/telemetry/history	Last 200 events as JSON
GET	/mesh/health/agents	Per-agent health scores

---

Environment Variables

Variable	Default	Description
MESH_VAULT_BACKEND	sqlite	sqlite or redis
MESH_VAULT_PATH	./mesh_vault.db	SQLite file path
MESH_REDIS_URL	redis://localhost:6379/0	Redis URL
MESH_PACKET_TTL_S	86400	Packet TTL (24h default)

---

Audit Status — v0.1.0

Check	Status
Pydantic v3 compatibility	✅ Pass
All async — zero blocking calls	✅ Pass
SQLite WAL mode (concurrent writes)	✅ Fixed
schema_version + trace_id in StatePacket	✅ Fixed
kwargs sanitization — no metadata leakage	✅ Fixed
Infinite hot-swap loop prevention	✅ Fixed
functools.wraps on decorator	✅ Pass
Parameterized SQL — no injection risk	✅ Pass
SSE double newline format	✅ Pass
No circular imports	✅ Pass

---

Tech Stack (2026)

FastAPI + LangGraph · PydanticAI · MCP 1.0 · SQLite/Redis · SSE · Rich · GitLab CI

---

License

MIT — owned by no one, available to everyone.

Built by AgenThink · Kuwait City · ADGM-Registered

The open protocol for AI agent discovery, state portability, and real-time routing.