Nuvu Agent

A developer-first Python framework for building agentic workflows with multi-provider LLM support. Removes boilerplate while standardizing how LLMs execute functions and connect to external data via MCP.

What It Does

Register Python functions as tools with a decorator, connect to remote MCP servers, and run an agent loop — the framework handles provider differences, tool schema conversion, and execution automatically.

Your Code → NuvuAgent → (OpenAI | Anthropic | Gemini) → Your Tools → Response

Installation

# Install from source
pip install -e ".[all,dev]"

# Core (includes OpenAI support)
pip install nuvu-agent

# With Anthropic (Claude) support
pip install nuvu-agent[anthropic]

# With Google Gemini support
pip install nuvu-agent[gemini]

# With FastAPI integration
pip install nuvu-agent[fastapi]

# Everything
pip install nuvu-agent[all]

Quick Start

from nuvu_agent import NuvuAgent, tool
from nuvu_agent.tools import NuvuMCPTool

# 1. Register a local tool with a decorator
@tool
def calculate_discount(price: float, percentage: float) -> float:
    """Calculates final price after applying a discount percentage."""
    return price * (1 - percentage / 100)

# 2. Connect to a remote MCP server (optional)
nuvu_mcp = NuvuMCPTool(endpoint="https://mcp.nuvu.dev")

# 3. Create an agent — swap the model string to change provider
agent = NuvuAgent(
    model="claude-3-5-sonnet-20241022",  # or "gpt-4o" or "gemini-1.5-pro"
    tools=[calculate_discount, nuvu_mcp],
    system_prompt="You are a helpful data analyst using Nuvu tools.",
)

# 4. Run
response = agent.run("What is $150 with a 15% discount?")
print(response)

The provider is auto-detected from the model name. Set your API key via environment variable:

export ANTHROPIC_API_KEY="sk-ant-..."   # for claude-* models
export OPENAI_API_KEY="sk-..."          # for gpt-* / o1-* / o3-* models
export GOOGLE_API_KEY="..."             # for gemini-* models

---

Core Concepts

The @tool Decorator

Convert any typed Python function into an agent tool. Type hints become the JSON schema automatically:

from nuvu_agent import tool
from typing import Literal, Optional

@tool
def search_orders(
    customer: str,
    status: Literal["pending", "shipped", "delivered"],
    limit: int = 20,
) -> str:
    """Search orders by customer name and status."""
    # Your implementation here
    return f"Found orders for {customer} with status {status}"

@tool
async def fetch_price(symbol: str) -> float:
    """Fetch the current price for a stock symbol."""
    # Async functions work too
    ...

Supported types: str, int, float, bool, list[X], Optional[X], Literal[...].

The decorator:

• Extracts function name → tool name
• Extracts docstring → tool description
• Maps type hints → JSON Schema parameters
• Wraps sync functions for async execution
• Validates arguments at runtime via Pydantic v2

NuvuMCPTool (Remote MCP Tools)

Connect to any MCP-compliant server to instantly expose its tools:

from nuvu_agent.tools import NuvuMCPTool

# Connects to MCP server, discovers tools via JSON-RPC handshake
mcp = NuvuMCPTool(
    endpoint="https://mcp.nuvu.dev",
    auth_token="optional-bearer-token",
)

# Pass to agent — all remote tools are available
agent = NuvuAgent(model="gpt-4o", tools=[mcp])

The MCP bridge:

• Performs the MCP initialize + tools/list handshake
• Converts remote schemas to LLM-compatible function definitions
• Routes execution back to the server via tools/call
• Caches tool discovery (handshake happens once)

Multi-Provider Support

The same code works across providers — just change the model string:

# OpenAI
agent = NuvuAgent(model="gpt-4o", tools=[...])

# Anthropic (Claude)
agent = NuvuAgent(model="claude-3-5-sonnet-20241022", tools=[...])

# Google Gemini
agent = NuvuAgent(model="gemini-1.5-pro", tools=[...])

# Explicit provider (for custom endpoints or ambiguous model names)
agent = NuvuAgent(model="my-custom-model", provider="openai", tools=[...])

Provider detection prefixes:

Prefix	Provider
claude-*	Anthropic
gpt-*, o1-*, o3-*, o4-*	OpenAI
gemini-*	Google Gemini

Agent Loop

NuvuAgent.run() executes a standard agent loop:

User Prompt → LLM → [Tool Call → Execute → Result]* → Final Response

• Loops up to max_iterations times (default: 10)
• Tool errors are caught and fed back as observations (never crashes)
• Supports both sync (run()) and async (arun()) execution

# Sync
response = agent.run("Analyze this data")

# Async
response = await agent.arun("Analyze this data")

---

Advanced: Server Integration (FastAPI)

For production deployments with streaming SSE, sessions, and multi-tenant auth, use the lower-level AgentOrchestrator:

from nuvu_agent import AgentConfig, ToolRegistry, APIToolExecutor
from nuvu_agent.session import InMemorySessionStore
from nuvu_agent.integrations.fastapi import create_agent_router
from fastapi import FastAPI

config = AgentConfig()  # reads AGENT_* env vars

registry = ToolRegistry()
registry.register({
    "type": "function",
    "function": {
        "name": "search_orders",
        "description": "Search orders by customer name or status",
        "parameters": {
            "type": "object",
            "properties": {
                "customer": {"type": "string"},
                "status": {"type": "string", "enum": ["pending", "shipped", "delivered"]},
            },
            "required": []
        }
    }
}, category="read")

executor = APIToolExecutor(api_base_url="http://localhost:8000")

@executor.handler("search_orders")
async def handle_search(args, context):
    params = {k: v for k, v in args.items() if v}
    return await executor.api_get("/api/orders", params=params, context=context)

class OrderPrompt:
    def build(self, user_id, context=None):
        return "You are an order management assistant. Use tools to find real data."

app = FastAPI()
router = create_agent_router(
    config=config,
    tool_registry=registry,
    tool_executor=executor,
    session_store=InMemorySessionStore(),
    prompt_builder=OrderPrompt(),
)
app.include_router(router, prefix="/api/agent")

---

Skills (Progressive Disclosure)

Skills are modular instruction sets that load on-demand — the agent only pays the token cost when it activates a skill. This works identically across all providers since skills are injected into the system prompt.

Level	When Loaded	What	Token Cost
1. Metadata	Always (system prompt)	Name + description + trigger	~100 tokens/skill
2. Instructions	On demand (load_skill)	Full SKILL.md body	1-5K tokens
3. Resources	On demand (read_skill_file)	Additional files	As needed

skills/
├── weather-analyzer/
│   ├── SKILL.md              # Frontmatter + instructions
│   └── REFERENCE.md          # Additional reference (Level 3)
└── data-analyzer/
    └── SKILL.md

from nuvu_agent.skills import SkillRegistry

skill_registry = SkillRegistry(skills_root="./skills")

# Include skill metadata in system prompt
system_prompt = "You are a helpful assistant.\n\n" + skill_registry.system_prompt_section()

---

Knowledge Base

Index markdown documentation for domain Q&A — no vector database needed:

from nuvu_agent.knowledge import MarkdownKnowledgeBase

kb = MarkdownKnowledgeBase(docs_root="./docs")
results = kb.search("shipping policy", max_results=3)

---

Environment Variables

Variable	Default	Description
OPENAI_API_KEY	—	API key for OpenAI models
ANTHROPIC_API_KEY	—	API key for Anthropic models
GOOGLE_API_KEY	—	API key for Gemini models
AGENT_LLM_API_KEY	—	API key for FastAPI orchestrator
AGENT_LLM_BASE_URL	https://api.openai.com/v1	LLM endpoint (orchestrator)
AGENT_LLM_MODEL	gpt-4o	Model (orchestrator)
AGENT_MAX_TOKENS	4096	Max tokens per response
AGENT_MAX_TOOL_ROUNDS	10	Max tool-calling iterations
AGENT_TEMPERATURE	0.0	LLM temperature

---

Architecture

┌─────────────────────────────────────────────────────────────────┐
│  Your Application                                               │
└───────────────────────────────┬─────────────────────────────────┘
                                │
                ┌───────────────┴───────────────┐
                │        nuvu-agent             │
                │                               │
                │  ┌─────────────────────────┐  │
                │  │      NuvuAgent          │  │
                │  │   (simple top-level)    │  │
                │  └────────────┬────────────┘  │
                │               │               │
                │  ┌────────────▼────────────┐  │
                │  │   Provider Adapters     │  │
                │  │  ┌───────┬──────┬────┐  │  │
                │  │  │OpenAI │Anthr.│Gem.│  │  │
                │  │  └───────┴──────┴────┘  │  │
                │  └────────────┬────────────┘  │
                │               │               │
                │  ┌────────────▼────────────┐  │
                │  │    Tool Execution       │  │
                │  │  ┌────────┬─────────┐   │  │
                │  │  │ @tool  │NuvuMCP  │   │  │
                │  │  │(local) │(remote) │   │  │
                │  │  └────────┴─────────┘   │  │
                │  └─────────────────────────┘  │
                └───────────────────────────────┘

---

OpenAI-Compatible Providers

Any service that exposes an OpenAI-compatible API (vLLM, Ollama, LiteLLM, Together AI, Groq, Azure OpenAI, etc.) works out of the box — pass base_url and set provider="openai":

# Local Ollama
agent = NuvuAgent(
    model="llama3",
    provider="openai",
    base_url="http://localhost:11434/v1",
)

# Together AI
agent = NuvuAgent(
    model="meta-llama/Llama-3-70b-chat-hf",
    provider="openai",
    api_key="your-together-key",
    base_url="https://api.together.xyz/v1",
)

# Groq
agent = NuvuAgent(
    model="llama-3.1-70b-versatile",
    provider="openai",
    api_key="your-groq-key",
    base_url="https://api.groq.com/openai/v1",
)

# Azure OpenAI
agent = NuvuAgent(
    model="gpt-4o",
    provider="openai",
    api_key="your-azure-key",
    base_url="https://your-resource.openai.azure.com/openai/deployments/gpt-4o/",
)

The provider="openai" override bypasses model-name auto-detection, and base_url routes requests to your endpoint.

---

Extending: Custom Providers

To add a new LLM provider, subclass BaseProvider and implement two methods:

from nuvu_agent.providers.base import BaseProvider
from nuvu_agent.schema import CanonicalTool, ProviderResponse, ToolCall, Usage

class MyCustomProvider(BaseProvider):
    def __init__(self, model: str, api_key: str | None = None, **kwargs):
        super().__init__(model, api_key, **kwargs)
        # Initialize your SDK client here

    def format_tools(self, tools: list[CanonicalTool]) -> list[dict]:
        """Convert canonical tools to your provider's wire format."""
        return [
            {
                "name": t.name,
                "description": t.description,
                "parameters": t.parameters_json_schema(),
            }
            for t in tools
        ]

    async def complete(self, messages, tools, *, temperature=0.0, max_tokens=4096) -> ProviderResponse:
        """Call your LLM and return a unified ProviderResponse."""
        # 1. Format tools and messages for your API
        # 2. Make the API call
        # 3. Parse response into ProviderResponse

        return ProviderResponse(
            content="response text",
            tool_calls=[],  # list of ToolCall(id, name, arguments)
            stop_reason="end_turn",  # or "tool_use"
            usage=Usage(prompt_tokens=0, completion_tokens=0),
        )

Then pass the instance directly to NuvuAgent via the provider parameter:

from nuvu_agent import NuvuAgent, tool

@tool
def greet(name: str) -> str:
    """Say hello."""
    return f"Hello, {name}!"

provider = MyCustomProvider(model="my-model", api_key="...")
agent = NuvuAgent(
    model="my-model",
    provider=provider,  # pass instance directly — bypasses auto-detection
    tools=[greet],
)
response = agent.run("Greet Alice")

The ProviderResponse contract is simple:

• content: text output (or None if only tool calls)
• tool_calls: list of ToolCall(id=str, name=str, arguments=dict)
• stop_reason: "end_turn" (done) or "tool_use" (wants to call tools)
• usage: optional token counts

---

License

Apache License, Version 2.0