Build LLM agents in Python. Any model. Any tool. No magic.

Quick Start | Examples | Docs | Why Cyclops

---

Cyclops is a thin wrapper around LiteLLM that gives you agents, tool calling, streaming, structured output, sessions, and MCP in one clean API. It works with every model LiteLLM supports. It doesn't hide the underlying calls from you.

Install

pip install cyclops-ai

uv add cyclops-ai

Quick Start

from cyclops import Agent, AgentConfig

agent = Agent(AgentConfig(model="gpt-4o-mini"))
print(agent.run("What is the capital of Japan?"))

Add tools:

from cyclops import Agent, AgentConfig
from cyclops.toolkit import tool

@tool
def get_price(ticker: str) -> str:
    """Get stock price for a ticker symbol"""
    return f"{ticker}: $142.50"

agent = Agent(AgentConfig(model="gpt-4o-mini"), tools=[get_price])
print(agent.run("What is Apple's stock price?"))
# The agent calls get_price("AAPL") automatically

What's included

Tool loop	Calls tools in a loop until the model stops asking for them, not just one round
Subagents	agent.as_tool() lets one agent delegate tasks to others, with shared cancellation, events and budgets
Streaming	agent.stream() and agent.astream() for token-by-token output
Structured output	agent.run(..., response_model=MyModel) returns a Pydantic instance
Images and files	Run input can be OpenAI chat content parts: images, PDFs and audio, checked against what the model takes
Cost tracking	agent.run_with_response() returns tokens used and estimated cost
Skills	Load instructions from SKILL.md folders only when the task needs them
Sessions	Session transcripts with usage, saved and loaded through a SessionStore (in memory or files), automatically after each step or run; forked and truncated
MCP	Connect to any MCP server as a tool source, or expose your tools as an MCP server
Plugins	Install cyclops-toolkit-* packages and tools are discovered automatically
Any LLM	OpenAI, Anthropic, Groq, Gemini, Ollama, Together AI, Bedrock, and 100+ more via LiteLLM
Provider options	completion_kwargs passes any LiteLLM argument (keys, headers, retries); typed reasoning_effort and thinking_budget; reasoning stays in history; prompt_caching on by default
Fallback routing	Pass a LiteLLM Router for automatic failover and load balancing
Naive mode	Prompt-based tool calling for models without native function calling support

Streaming

agent = Agent(AgentConfig(model="gpt-4o-mini"))

for chunk in agent.stream("Explain the water cycle"):
    print(chunk, end="", flush=True)

Async:

async for chunk in agent.astream("Write a haiku about Python"):
    print(chunk, end="", flush=True)

Structured Output

from pydantic import BaseModel
from cyclops import Agent, AgentConfig

class Summary(BaseModel):
    title: str
    key_points: list[str]
    sentiment: str

agent = Agent(AgentConfig(model="gpt-4o-mini"))
result = agent.run("Summarize the French Revolution", response_model=Summary)

print(result.title)       # "The French Revolution"
print(result.key_points)  # ["Storming of the Bastille", ...]

Cost and Token Tracking

response = agent.run_with_response("Write a product description for noise-cancelling headphones")

print(f"Model:      {response.model}")
print(f"Tokens:     {response.tokens_used}")
print(f"Cost:       ${response.cost:.6f}")
print(f"Answer:     {response.content}")

Multi-Turn Conversations

Each run starts from a fresh transcript. A Session carries a conversation from one run to the next:

from cyclops import Session

agent = Agent(AgentConfig(model="gpt-4o-mini"))
session = Session()

agent.run("My name is Alice and I work in Tokyo", session=session)
print(agent.run("Where do I work?", session=session))  # "You work in Tokyo"

print(agent.run("What's my name?"))  # no session, no memory of Alice

To save a conversation and continue it later, keep its session in a SessionStore:

from cyclops import FileSessionStore, Session

store = FileSessionStore("./sessions")
session = await store.load("alice") or Session(id="alice")
await agent.arun("Where do I work?", session=session)
await store.save(session)

Or give the agent the store, Agent(config, session_store=store), and every run with a session saves it as it goes. session.fork() copies a session under a new id, and session.truncate(index) goes back to an earlier message.

MCP

Connect to MCP servers over stdio, streamable HTTP or SSE and give their tools to an agent:

from cyclops.mcp import MCPClient

async with MCPClient() as mcp:
    await mcp.attach("files", ["npx", "-y", "@modelcontextprotocol/server-filesystem", "."])
    await mcp.attach("docs", url="https://example.com/mcp", headers={"Authorization": "Bearer ..."})

    agent = Agent(config, tools=mcp.tools())  # files__read_file, docs__search, ...
    response = await agent.arun("Summarize README.md")

Expose your tools as an MCP server:

from cyclops.mcp import MCPServer
from cyclops.toolkit import tool

@tool
def lookup(id: str) -> str:
    """Look up a record by ID"""
    return f"Record {id}: active"

server = MCPServer("my-server")
server.add_tool(lookup)
await server.run_stdio()

Observability

Wire TelemetryHooks to get OpenTelemetry spans for every LLM call and tool execution. Works with any OTLP backend — Jaeger, Honeycomb, Grafana Tempo, Datadog, or the console.

from cyclops import Agent, AgentConfig, TelemetryHooks

agent = Agent(AgentConfig(model="groq/llama-3.1-8b-instant", hooks=TelemetryHooks.console()))
agent.run("What files are in the current directory?")

Send to Jaeger / Tempo / Datadog instead:

# uv add opentelemetry-exporter-otlp-proto-grpc
agent = Agent(AgentConfig(model="groq/llama-3.1-8b-instant", hooks=TelemetryHooks.otlp("http://localhost:4317")))
agent.run("What files are in the current directory?")

Span hierarchy per run: invoke_agent with chat (token counts) and execute_tool <name> children, using the OpenTelemetry GenAI semantic conventions. The factories build their own tracer provider; pass TelemetryHooks(tracer_provider=...) to use yours. One instance is safe to share across concurrent runs. See Observability guide for OTLP/Jaeger setup and full attribute reference.

Providers

Switch models by changing one string. Set the matching API key as an environment variable.

# OpenAI
Agent(AgentConfig(model="gpt-4o-mini"))                        # OPENAI_API_KEY

# Anthropic
Agent(AgentConfig(model="claude-haiku-4-5-20251001"))          # ANTHROPIC_API_KEY

# Groq (fast, free tier)
Agent(AgentConfig(model="groq/llama-3.1-8b-instant"))          # GROQ_API_KEY

# Ollama (local, no API key)
Agent(AgentConfig(model="ollama/qwen3:4b"))

# Google
Agent(AgentConfig(model="gemini/gemini-1.5-flash"))            # GEMINI_API_KEY

Fallback Routing

from litellm import Router
from cyclops import Agent, AgentConfig

router = Router(
    model_list=[
        {"model_name": "primary", "litellm_params": {"model": "gpt-4o-mini"}},
        {"model_name": "primary", "litellm_params": {"model": "groq/llama-3.1-8b-instant"}},
    ],
    fallbacks=[{"primary": ["groq/llama-3.1-8b-instant"]}],
    num_retries=2,
)

agent = Agent(AgentConfig(model="primary", router=router))

Why Cyclops?

Most agent frameworks add heavy abstractions, require specific clouds, or make it hard to see what's actually being sent to the LLM. Cyclops doesn't. The Agent class is ~400 lines. A session's messages is a plain list of dicts in LiteLLM format. Every call goes straight to LiteLLM.

	Cyclops	LangChain	smolagents
Lines to first agent	3	15+	8
Any LiteLLM model	yes	partial	yes
MCP native	yes	plugin only	no
Streaming	yes	yes	no
Structured output	yes	yes	no
Cost tracking	yes	partial	no
Abstraction	thin	thick	medium

Examples

See the examples/ directory:

File	What it shows
basic_agent.py	Hello world, multi-turn
agent_with_tools.py	Tool calling
streaming_example.py	stream() and astream()
structured_output.py	response_model with Pydantic
cost_tracking.py	Tokens and cost
multi_turn_agent.py	Conversations with Session
tool_loop_demo.py	Multi-step tool chains
skills_example.py	Skills from SKILL.md folders
async_agent.py	arun() and concurrent calls
subagents.py	Agents as tools with as_tool()
different_llms.py	Groq, Ollama, OpenAI, Together AI
router_fallback.py	Automatic failover
mcp_server.py	Expose tools via MCP
plugin_system.py	Auto-discover toolkit packages
otel_example.py	OpenTelemetry tracing

Development

git clone https://github.com/gopaljigaur/cyclops
cd cyclops
uv sync
uv run pre-commit install
uv run pytest

Soak tests repeat each kind of run a thousand times against mocked completions and fail when memory, objects, tasks, threads, open files or child processes pile up. They are left out of uv run pytest; run them with uv run pytest -m soak -n auto. CI runs them nightly, and on a pull request once it has the soak label.

Contributing

Bug reports, feature requests, and pull requests are welcome. Open an issue first for large changes.

License

MIT