Cogent

Five concepts. One loop. 10K lines. 13 providers. No vendor SDKs.

Cogent is a Python framework for building AI agents that you can actually ship. Small, typed, composable — one runner loop, one middleware protocol, one tool decorator, raw HTTP to every provider.

v2.0.0 is now the main branch — a ground-up rewrite that replaced 77,000 lines and 7 runtime dependencies with a 10,000-line core and 4 dependencies, while shipping more capabilities. See the release notes for the full story. The 1.x codebase lives on the v1 branch. There are no compatibility shims — pin cogent-ai<2 if you need 1.x.

---

Install

pip install "cogent-ai[openai]"     # anthropic, ollama, voyage, sqlite, otel, rich, all

Cogent targets Python 3.13+. Four runtime dependencies: pydantic, httpx, anyio, python-dotenv. Zero vendor SDKs — all 13 providers speak HTTP directly.

Hello, agent

import asyncio
from cogent import Agent

agent = Agent(
    model="openai/gpt-4o-mini",
    instructions="You are a Kyoto travel guide. Keep answers short and concrete.",
)

async def main() -> None:
    result = await agent.run("Where should I eat tofu in Kyoto?")
    print(result.value)

asyncio.run(main())

A tool, a structured output

from pydantic import BaseModel
from cogent import Agent, tool

class Recipe(BaseModel):
    name: str
    steps: list[str]

@tool
async def lookup(name: str) -> str:
    """Look up a recipe."""
    return "miso soup: warm dashi, dissolve miso, drop in tofu and wakame."

agent = Agent(
    model="openai/gpt-4o-mini",
    tools=(lookup,),
    output=Recipe,
)
recipe = (await agent.run("recipe for miso soup")).value

Streaming

async for event in agent.run("describe spring in Kyoto", stream=True):
    print(event)

Middleware (Retry · CostBudget · Cache · Approval · …)

from cogent import Agent, Approval, CostBudget

agent = Agent(
    model="openai/gpt-4o-mini",
    middleware=(
        CostBudget(limit_usd=1.00),
        Approval(predicate=lambda call: call.name == "send_email"),
    ),
)

Retry is built in (retries=3); set retries=0 to disable or pass a custom Retry(...) in middleware. Middleware composes from left to right around every model_call and tool_call. Approvals turn into a Paused result you can resume from later, in-process or across processes (set Agent(store=…) and open with agent.thread(id=…)).

Sub-agents

from cogent import Agent, as_tool, sequential

researcher = Agent(model=..., instructions="Find sources.")
writer = Agent(model=..., instructions="Draft the brief.")

# Use one agent as a tool inside another
team = Agent(model=..., tools=(as_tool(researcher), as_tool(writer)))

# Or run them sequentially, threading each output as the next input
result = await sequential(researcher, writer, task="Outline 3 days in Kyoto.")

Persistence + resume

from cogent import Agent, DictStore
# from cogent.store import SQLiteStore  # durable, FTS5-backed

store = DictStore()
agent = Agent(model="openai/gpt-4o-mini", store=store)

thread = agent.thread(id="trip-1")
result = await thread.send("Plan a 3-day trip.")
# ... process exits ...

# Later — in another process, with the same store backend:
resumed = agent.thread(id="trip-1")
final = await resumed.send("Make day 2 vegetarian.")

store= covers both thread snapshots and recall/remember. Pass threads= or memory= only when you want different backends for each.

Testing without the network

from cogent import Agent
from cogent.core.messages import ToolCall
from cogent.testing import MockModel, MockTool, expect, reply, tool_use

model = MockModel(
    responses=[
        tool_use(ToolCall(id="c1", name="fetch", args={"city": "Kyoto"})),
        reply("Kyoto is sunny."),
    ]
)
fetch = MockTool(name="fetch", returns="sunny", arg_fields={"city": str})

agent = Agent(model=model, tools=(fetch,))
result = await agent.run("how's Kyoto?")
expect(result).to_be_done().with_value_containing("sunny").with_steps(2)
assert fetch.calls == [{"city": "Kyoto"}]

Observability

from cogent import Agent
from cogent.middleware import JSONLogger, Logger
# OTelTracer / OTelMetrics are exposed lazily on cogent.middleware when
# the 'otel' extra is installed.

agent = Agent(
    model=...,
    middleware=(Logger(), JSONLogger(redact=lambda k, _: "secret" in k.lower())),
)

What's in the box

Area	Modules
Core	Agent, Thread, Context, Message, Result, Event
Models	OpenAI, Anthropic, Ollama, Groq, DeepSeek, Gemini, Mistral, Cohere, xAI, OpenRouter, Together, Cerebras, Cloudflare; embedders for OpenAI, Voyage, Ollama
Tools	@tool, ToolBase, builtins (recall, remember, update_block)
Middleware	Retry, RateLimit, CostBudget, Cache, Approval, Compact, AutoRecall, Logger, JSONLogger, OTelTracer, OTelMetrics
Store	DictStore, SQLiteStore (FTS5) — thread snapshots, recall, episodes
Composition	as_tool, sequential, until, replay
Testing	MockModel, MockTool, expect

Documentation

• Getting Started
• Concepts — the five building blocks
• Middleware — retry, rate limiting, cost, caching, approval, observability
• Memory — recall, remember, episodic
• Testing — mock models, assertions
• Examples — runnable notebooks

License

MIT — see LICENSE.