AgentKit

Build observable Python agents that use tools, remember context, delegate work, and ship behind a secure API.

AgentKit is a lightweight, transparent framework for ReAct-style agents. It gives you the primitives you need to move from a prompt demo to a shippable agent system: provider adapters, typed tool schemas, streaming, structured output, memory, multi-agent routing, OpenAPI tool import, observability hooks, and a FastAPI server with production-minded guardrails.

Why AgentKit

Most agent frameworks either hide the loop or make you assemble too much plumbing. AgentKit keeps the loop visible and the surface area small.

Need	What AgentKit gives you
Tool-using agents	ReAct loop, typed tool registry, parallel action parsing, built-in tools
Typed outputs	Pydantic response models with graceful fallback on parse failure
Memory	Sliding short-term context plus optional vector memory backends
Multi-agent systems	Manager/team delegation and swarm-style transfers
API tools	OpenAPI 3.x and Swagger 2 import with auth injection
Production API	API key auth, model allowlist, rate limiting, timeouts, body-size limits
Observability	Step traces, token accounting, cost estimates, Langfuse-ready hooks

Install

pip install agentkit-ai

Install only the optional capabilities you need:

pip install "agentkit-ai[memory]"       # ChromaDB + sentence-transformers
pip install "agentkit-ai[integrations]" # GitHub + Notion
pip install "agentkit-ai[e2b]"          # E2B code interpreter sandbox
pip install "agentkit-ai[qdrant]"
pip install "agentkit-ai[pinecone]"
pip install "agentkit-ai[weaviate]"
pip install "agentkit-ai[pgvector]"
pip install "agentkit-ai[all]"

Python 3.10, 3.11, and 3.12 are tested in CI.

Quickstart

import asyncio

from agentkit.agent import Agent
from agentkit.llm.openai import OpenAILLM
from agentkit.memory import ShortTermMemory
from agentkit.tools import ToolRegistry, tool


@tool
def get_weather(city: str, unit: str = "C") -> str:
    """Return example weather for a city."""
    return f"22{unit} and sunny in {city}"


async def main() -> None:
    agent = Agent(
        llm=OpenAILLM(model_name="gpt-4o-mini"),
        tools=ToolRegistry([get_weather]),
        memory=ShortTermMemory(),
        system_prompt="You are concise, helpful, and transparent.",
    )

    response = await agent.run("What is the weather in Istanbul?")

    print(response.final_answer)
    print(response.total_tokens, response.estimated_usd)
    for step in response.steps:
        print(step.step_type, step.tool_name, step.tool_input, step.content)


asyncio.run(main())

Architecture

flowchart LR
    User["User or app"] --> Agent["Agent ReAct loop"]
    Agent --> Memory["Short-term memory"]
    Agent --> LLM["LLM provider"]
    Agent --> Registry["Tool registry"]
    Registry --> Builtins["Built-in tools"]
    Registry --> OpenAPI["Imported OpenAPI tools"]
    Registry --> Sandbox["Sandboxed execution"]
    Agent --> Response["AgentResponse"]
    Response --> Steps["Traceable steps"]
    Response --> Cost["Token and cost accounting"]

    Team["Team manager"] --> Agent
    Team --> Specialists["Specialist agents"]
    Swarm["Swarm router"] --> Specialists

    API["FastAPI server"] --> Agent
    API --> Guardrails["Auth, allowlist, rate limit, timeout"]

Loading

Live Console

The bundled FastAPI server includes a minimal live console for sending tasks, streaming answers, and inspecting the auditable agent trace returned by the API.

Core Features

Streaming without leaking chain-of-thought

AgentKit buffers each LLM iteration until it knows whether the output is a tool call or a final response. End users receive the final answer stream; internal Thought and Action text stays in the agent trace.

async for chunk in agent.arun("Explain Python in one sentence."):
    print(chunk, end="", flush=True)

Structured output

from pydantic import BaseModel


class Profile(BaseModel):
    name: str
    skills: list[str]


response = await agent.run("Describe Guido van Rossum.", response_model=Profile)
if response.structured_output:
    print(response.structured_output.name)

Tools with safety controls

Built-in tools:

• web_search
• local_python_repl
• sandbox_python_repl with the e2b extra
• read_file
• write_file

local_python_repl runs in an isolated Python subprocess with an AST allowlist and timeout. Imports, file/network access, dunder access, and arbitrary attributes are rejected. For untrusted or general-purpose code execution, use sandbox_python_repl.

Human approval can pause every registered tool call:

agent = Agent(
    llm=llm,
    tools=tools,
    memory=ShortTermMemory(),
    system_prompt="Ask before using tools.",
    require_human_approval=True,
)

Provider adapters

from agentkit.llm import AnthropicLLM, GroqLLM, OllamaLLM, OpenAILLM

openai_llm = OpenAILLM(model_name="gpt-4o-mini")
anthropic_llm = AnthropicLLM(model_name="claude-3-5-sonnet-20241022")
groq_llm = GroqLLM(model_name="llama-3.3-70b-versatile")
ollama_llm = OllamaLLM(model_name="llama3.2")

Provider API keys are read from OPENAI_API_KEY, ANTHROPIC_API_KEY, and GROQ_API_KEY.

Memory

Short-term memory keeps the active conversation inside a token budget:

from agentkit.memory import ShortTermMemory

memory = ShortTermMemory(max_tokens=4000)

Long-term memory uses Chroma by default and can accept any BaseVectorDB implementation:

from agentkit.memory import LongTermMemory

memory = LongTermMemory(
    collection_name="agent_memory",
    persist_directory="./chroma_db",
)
memory.add("AgentKit supports tools.", {"source": "docs"})
results = memory.search("What does AgentKit support?", k=3)

from agentkit.memory import LongTermMemory
from agentkit.memory.vector_db.qdrant import QdrantVectorDB

db = QdrantVectorDB(url="http://localhost:6333")
memory = LongTermMemory(db=db)

Available vector extras: memory, qdrant, pinecone, weaviate, and pgvector.

Multi-Agent Workflows

Team delegation

from agentkit.orchestrator import Team

manager = Agent(
    name="manager",
    llm=llm,
    tools=ToolRegistry(),
    memory=ShortTermMemory(),
    system_prompt="Delegate specialist work when useful.",
)

researcher = Agent(
    name="researcher",
    llm=llm,
    tools=ToolRegistry([web_search]),
    memory=ShortTermMemory(),
    system_prompt="Research and summarize.",
)

team = Team(manager)
team.add_agent("researcher", researcher)
response = await team.run("Research the latest Python packaging changes.")

delegate_to_agent is registered after manager construction. AgentKit refreshes tool schemas before every LLM iteration, so dynamically registered Team and Swarm tools are present in the system prompt.

Swarm transfers

from agentkit.orchestrator import Swarm

swarm = Swarm(starting_agent=manager, max_transfers=10)
swarm.add_agent(researcher)
response = await swarm.run("Route this task to the right expert and finish it.")

max_transfers prevents infinite handoff loops.

OpenAPI To Tools

Turn an API spec into callable agent tools:

from agentkit.tools.openapi import import_openapi
from agentkit.tools import ToolRegistry

tools = import_openapi(
    "https://api.example.com/openapi.json",
    auth={
        "ApiKeyAuth": "secret-api-key",
        "BearerAuth": "access-token",
    },
)

registry = ToolRegistry(tools)

The importer supports:

• OpenAPI 3.x and Swagger 2
• local $ref resolution
• path, query, header, and cookie parameters
• JSON, form, and multipart request bodies
• API key, HTTP Basic/Bearer, OAuth2, and OpenID Connect credentials
• path-level parameters and operation overrides

Credentials are supplied to import_openapi() and are not exposed in tool schemas.

Ship An API

Copy the environment template:

cp .env.example .env

Configure at least:

OPENAI_API_KEY=...
AGENTKIT_API_KEY=use-a-long-random-secret
AGENTKIT_ALLOWED_MODELS=gpt-4o,gpt-4o-mini
AGENTKIT_REDIS_URL=redis://redis:6379/0

Run the server:

agentkit deploy --host 0.0.0.0 --port 8000

Call it:

curl http://localhost:8000/api/chat \
  -H "Content-Type: application/json" \
  -H "X-API-Key: use-a-long-random-secret" \
  -d '{"message":"Hello","model":"gpt-4o-mini"}'

Bearer auth is also accepted:

Authorization: Bearer use-a-long-random-secret

API guardrails

• fail-closed API key authentication
• constant-time key comparison
• per-client rate limiting with Redis support for multi-instance deployments
• model allowlist
• request body size limit
• execution timeout for JSON and SSE endpoints
• no host code-execution tool in the server registry
• DOM rendering through text nodes rather than innerHTML

If AGENTKIT_REDIS_URL is set, rate limits are shared across processes and instances. Without Redis, AgentKit falls back to a process-local in-memory limiter for local development.

Docker

docker compose up --build

The production image:

• runs as a non-root user
• installs only core API dependencies
• uses /health for its healthcheck
• persists the default Chroma directory at /app/chroma_db
• ships with a Redis service in docker-compose.yml for shared rate limiting

Vector database extras are intentionally excluded from the API image. Build a specialized image if the deployed server needs them.

CLI

agentkit --help
agentkit tools list
agentkit run "Hello" --model gpt-4o-mini
agentkit chat --model gpt-4o-mini
agentkit deploy --host 127.0.0.1 --port 8000

python -m agentkit provides the same commands.

Checkpoints

agent.save_checkpoint("checkpoint.json")
agent.load_checkpoint("checkpoint.json")

Checkpoint files contain conversation history and token counters. Protect them if conversation data is sensitive.

Examples

Runnable examples live at the repository root:

• example_agent.py
• example_tools.py
• example_memory.py
• example_multi_agent.py
• example_swarm.py
• example_structured_output.py
• example_openapi.py
• example_sandbox.py
• example_llm.py

Development

poetry install -E memory
poetry run pytest
poetry run ruff check agentkit tests
poetry run mypy --strict agentkit tests
poetry run bandit -q -r agentkit
poetry build
docker build --check .

CI runs tests on Python 3.10, 3.11, and 3.12, plus linting, type checking, security scanning, package build validation, and Docker validation.

Long-term memory unit tests use an in-memory fake vector backend so the default test suite does not download embedding models. To run the real Chroma/SentenceTransformer integration tests:

AGENTKIT_RUN_MEMORY_INTEGRATION=1 poetry run pytest tests/test_memory.py tests/test_edge_cases.py

Release publishing uses PyPI Trusted Publishing through GitHub Actions OIDC, so the workflow does not require a long-lived PYPI_API_TOKEN secret.

License

MIT