nexus

🌀 Nexus

Multi-Agent Orchestration Framework for Python

3 orchestration modes · 100+ LLM providers · Built-in tool calling & MCP

English | 中文 | Examples

---

Why Nexus?

Most agent frameworks force you to choose: simple API (Swarm) or powerful workflows (LangGraph). Nexus gives you both — start with 5 lines of code, scale to complex DAG workflows, LLM-based routing, and self-organizing agent teams.

import asyncio
from nexus import Team

async def main():
    async with Team(model="gpt-4o", api_key="sk-...") as team:
        team.add_agent("coder", instructions="You write clean Python code.")
        team.add_agent("writer", instructions="You write engaging articles.")
        result = await team.run("Implement a binary search in Python")
        print(result)  # → Automatically routed to 'coder'

asyncio.run(main())

✨ Highlights

	Feature	Description
🔀	3 Orchestration Modes	Static Graph (DAG) · Dynamic Router (LLM triage) · Adaptive (embedding matching)
🌍	100+ LLM Providers	OpenAI, Anthropic, Groq, Deepseek, Mistral, Qwen, Ollama, Azure, Bedrock, Vertex AI… via LiteLLM
🔧	Tool Calling	ReAct loop (Think → Act → Observe → Repeat) with auto schema extraction from Python type hints
🔌	Protocol-Native	MCP for tool servers · A2A for agent discovery
💬	Multi-turn & Streaming	Conversation memory with chat() and real-time token streaming with stream()
🛡️	Production-Grade	Token budgets · Checkpoint/resume · Retry/fallback policies · Tracing · Metrics

---

Installation

# Recommended — unlocks 100+ LLM providers
pip install nexus-agents[litellm]

# Or pick a specific provider
pip install nexus-agents[openai]       # OpenAI only
pip install nexus-agents[anthropic]    # Anthropic only

# Everything (all providers + MCP + observability)
pip install nexus-agents[all]

---

Quick Start

Multi-Agent with Auto-Routing

import asyncio
from nexus import Team

async def main():
    async with Team(model="gpt-4o", api_key="sk-...") as team:
        team.add_agent("coder", instructions="You write Python code.")
        team.add_agent("writer", instructions="You write articles.")
        team.add_agent("analyst", instructions="You analyze data.")

        # The router examines each task and picks the best agent
        await team.run("Implement quicksort in Python")    # → coder
        await team.run("Write a blog post about AI agents") # → writer

asyncio.run(main())

Tool Calling (ReAct Loop)

Define tools as plain Python functions — Nexus extracts the JSON schema automatically:

def calculate(expression: str) -> str:
    """Evaluate a math expression. Example: '2**32 - 1'"""
    return str(eval(expression))

def get_current_time() -> str:
    """Get the current UTC time."""
    from datetime import datetime, timezone
    return datetime.now(timezone.utc).isoformat()

async with Team(model="gpt-4o", api_key="sk-...") as team:
    team.add_agent(
        "assistant",
        instructions="Use tools to answer questions accurately.",
        tools=[calculate, get_current_time],
    )
    result = await team.run("What is 2^32 - 1?")
    # 🔧 Agent calls calculate("2**32 - 1") → "4294967295"
    # ✨ Agent responds: "2³² - 1 = 4,294,967,295"

Multi-turn Conversation

async with Team(model="gpt-4o", api_key="sk-...") as team:
    team.add_agent("assistant", instructions="You are helpful.")

    r1 = await team.chat("What is the Fibonacci sequence?")
    r2 = await team.chat("Show me the first 10 numbers")  # Remembers context
    team.reset_conversation()  # Clear history

Streaming Output

async with Team(model="gpt-4o", api_key="sk-...") as team:
    team.add_agent("storyteller", instructions="You tell captivating stories.")

    async for chunk in team.stream("Tell me a story about a time-traveling robot"):
        print(chunk, end="", flush=True)

📂 More examples → See the examples/ directory for runnable demos: tool calling, MCP integration, Graph/Router/Adaptive orchestration modes, and more.

# Clone and run
git clone https://github.com/sontianye/nexus.git && cd nexus
pip install -e '.[all]'
cp examples/.env.example examples/.env  # Add your API key
python examples/demo_tools.py simple    # Try tool calling
python examples/demo_modes.py graph     # Try graph orchestration

---

Three Orchestration Modes

🔷 Graph — Static DAG Workflows

For deterministic pipelines where you know the exact steps:

from nexus.orchestration.graph import GraphOrchestrator, START, END

graph = GraphOrchestrator()
graph.add_node("research", research_fn)
graph.add_node("write", write_fn)
graph.add_node("review", review_fn)

graph.add_edge(START, "research")
graph.add_edge("research", "write")
graph.add_edge("write", "review")
graph.add_conditional_edge(
    "review",
    lambda state: "write" if state.get("needs_revision") else END
)

result = await graph.execute({"topic": "AI Agents"})

🔶 Router — Dynamic LLM-Based Routing

A triage agent examines each request and picks the best specialist:

from nexus.orchestration.topology import TopologyMode

async with Team(model="gpt-4o", mode=TopologyMode.ROUTER) as team:
    team.add_agent("coder", instructions="You write code.")
    team.add_agent("writer", instructions="You write prose.")
    await team.run("Implement quicksort in Python")  # → routes to coder

🔴 Adaptive — Embedding-Based Capability Matching

Agents are matched to tasks via cosine similarity on capability vectors — unique to Nexus:

from nexus import AgentCapability
from nexus.orchestration.topology import TopologyMode

async with Team(
    model="gpt-4o",
    mode=TopologyMode.ADAPTIVE,
    embedding_model="text-embedding-3-small",
) as team:
    team.add_agent("coder", capabilities=AgentCapability(coding=1.0, reasoning=0.8))
    team.add_agent("writer", capabilities=AgentCapability(creativity=1.0, language=0.9))
    await team.run("Write a poem about recursion")  # → capability match → writer

---

100+ LLM Providers

Switch models by changing one string. Zero code changes:

Team(model="gpt-4o")                                    # OpenAI
Team(model="anthropic/claude-sonnet-4-20250514")      # Anthropic
Team(model="groq/llama-3.1-70b-versatile")              # Groq (ultra-fast)
Team(model="deepseek/deepseek-chat")                     # Deepseek
Team(model="vertex_ai/gemini-1.5-pro")                   # Google Vertex AI
Team(model="bedrock/anthropic.claude-3-sonnet")          # AWS Bedrock
Team(model="ollama/llama3.1")                            # Local (Ollama)

# Any OpenAI-compatible endpoint
Team(model="openai/my-model", api_base="https://my-api.com/v1", api_key="sk-...")

# Mix models per agent
async with Team(model="gpt-4o") as team:
    team.add_agent("fast", model="groq/llama-3.1-70b-versatile")   # Speed
    team.add_agent("smart", model="anthropic/claude-sonnet-4-20250514") # Quality

Full list → LiteLLM Providers

---

MCP Protocol Integration

Connect agents to any MCP tool server:

from nexus.protocol.mcp import MCPToolProvider

# Local MCP server via stdio
async with MCPToolProvider("npx", ["-y", "@modelcontextprotocol/server-filesystem", "."]) as mcp:
    tools = await mcp.get_tools()
    team.add_agent("file_agent", tools=tools)

# Remote MCP server via HTTP/SSE
async with MCPToolProvider(url="https://mcp-server.example.com/sse") as mcp:
    tools = await mcp.get_tools()
    team.add_agent("web_agent", tools=tools)

---

Architecture

┌─────────────────────────────────────────────────────┐
│  Developer API                                       │
│  Team · CLI · @tool decorator · YAML config          │
├─────────────────────────────────────────────────────┤
│  Observability                                       │
│  Distributed Tracing · Metrics · Token Accounting    │
├─────────────────────────────────────────────────────┤
│  Orchestration                                       │
│  🔷 Graph  ·  🔶 Router  ·  🔴 Adaptive             │
├─────────────────────────────────────────────────────┤
│  Runtime                                             │
│  Execution Engine · Checkpoint · Retry / Fallback    │
│  Token Budget · Auto-Degradation Strategies          │
├─────────────────────────────────────────────────────┤
│  Protocol + Core                                     │
│  Agent · Task · Handoff · Message                    │
│  A2A Protocol · MCP Protocol · Discovery Service     │
├─────────────────────────────────────────────────────┤
│  Model Providers                                     │
│  OpenAI · Anthropic · Ollama · LiteLLM (100+)       │
└─────────────────────────────────────────────────────┘

Project Structure

nexus/
├── api.py                  # Team — high-level API
├── cli.py                  # CLI: init / run / chat / inspect
├── core/                   # Agent, Task, Handoff, Message, State
├── models/                 # OpenAI, Anthropic, Ollama, LiteLLM providers
├── orchestration/          # Graph, Router, Adaptive, TopologyEngine
├── protocol/               # A2A, MCP, MessageBus, Discovery
├── runtime/                # Engine, Checkpoint, Budget, Policies
└── observability/          # Tracing, Metrics

---

How Nexus Compares

Feature	Nexus	LangGraph	CrewAI	Swarm
Graph Workflows	✅	✅	—	—
Dynamic LLM Routing	✅	—	⚠️	✅
Adaptive Matching	✅	—	—	—
100+ Providers (LiteLLM)	✅	—	—	—
MCP Protocol	✅	—	—	—
A2A Protocol	✅	—	—	—
Tool Calling	✅	✅	✅	✅
Streaming	✅	✅	—	—
Token Budgets	✅	—	—	—
Checkpoint / Resume	✅	✅	—	—
Minimal API (5 lines)	✅	—	✅	✅

Note: "—" means not built-in. Some features may be available via plugins or custom code.

---

Roadmap

• Core primitives (Agent + Task + Handoff)
• 3 orchestration modes (Graph / Router / Adaptive)
• LiteLLM integration (100+ providers)
• Tool calling with ReAct loop
• MCP protocol integration
• A2A protocol support
• Streaming responses
• Multi-turn conversation
• Token budget management
• Checkpoint / resume
• Observability (tracing + metrics)
• CLI (init / run / chat / inspect)
• Agent memory (long-term RAG)
• OpenTelemetry export
• Web UI dashboard
• Multi-modal (vision, audio)
• Distributed execution

---

Contributing

Contributions are welcome! See CONTRIBUTING.md for guidelines.

git clone https://github.com/sontianye/nexus.git
cd nexus
uv venv --python 3.12 .venv && uv pip install --python .venv/bin/python -e ".[all,dev]"
.venv/bin/python tests/test_smoke.py  # Verify everything works

License

MIT © 2025–2026 Nexus Contributors

---

⭐ If Nexus is useful to you, please star this repo — it helps others discover it!

Examples · Issues · Discussions