miniAgent

A lightweight, production-ready orchestration SDK for multi-agent AI pipelines.

---

Table of Contents

• Why miniAgent?
• Installation
• Quick Start
• Configuration
  • YAML Configuration
  • ModelConfig Reference
  • Programmatic Configuration
• Core Concepts
• Agents
  • LiteLLMAgent
  • AnthropicAgent
  • GoogleADKAgent
  • Building Custom Agents
• Workflows
  • SequentialWorkflow
  • GraphWorkflow
  • ParallelWorkflow
  • LoopWorkflow
  • WorkflowRegistry
• Tools
  • Built-in Tools
  • Building Custom Tools
• LLM Providers
  • LiteLLMProvider
  • AnthropicProvider
  • Extended Thinking
  • Provider Registry
• State & Messages
• REST API
  • Running the Server
  • Agent Endpoints
  • Workflow Endpoints
  • Streaming (SSE)
  • Meta Endpoints
• Adapter Registry
• Exception Hierarchy
• Architecture
• Examples
• Testing
• Development

---

Why miniAgent?

Most agent frameworks bind you to a single vendor. miniAgent separates concerns cleanly:

Concern	miniAgent solution
LLM access	Pluggable providers — LiteLLM (80+ models) or native Anthropic SDK
Agent logic	BaseAgent interface; swap implementations without touching orchestration
Orchestration	Four workflow patterns covering every common use-case
Configuration	Declarative YAML with Pydantic validation at every boundary
Deployment	FastAPI REST layer included; SSE streaming out of the box
Observability	Typed exception hierarchy; structured logging throughout

---

Installation

# Core (LiteLLM + FastAPI included)
pip install miniAgent

# Or from source
pip install -e .

Optional extras

# Native Anthropic SDK (extended thinking, prompt caching)
pip install "miniAgent[anthropic]"

# Google ADK agent
pip install "miniAgent[google]"

# Everything including dev tools
pip install "miniAgent[all]"

Community adapters — Reference implementations for CrewAI and LangChain are available in examples/advanced/ as copy-paste starting points.

Requirements: Python ≥ 3.14

---

Quick Start

1. Set your API keys

export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."
export GOOGLE_API_KEY="AIza..."

2. Define agents in YAML

# agents.yaml
agents:
  - name: researcher
    system_prompt: "You are a thorough research assistant. Always cite sources."
    model:
      litellm_model: "openai/gpt-4o"
      temperature: 0.3
    max_iterations: 10

  - name: writer
    system_prompt: "You are a professional technical writer."
    model:
      litellm_model: "openai/gpt-4o"
      temperature: 0.8
    max_iterations: 5

3. Run a sequential pipeline

import asyncio
from miniAgent.core.config import AgentConfigLoader
from miniAgent.core.models.message import Message, MessageRole
from miniAgent.core.models.state import State
from miniAgent.orchestrator.sequential import SequentialWorkflow
from miniAgent.adapters.llm.litellm_agent import LiteLLMAgent
from miniAgent.providers.litellm_provider import LiteLLMProvider

# Load profiles and create agents
profiles = AgentConfigLoader.load_from_file("agents.yaml")
provider = LiteLLMProvider()
agents = {p.name: LiteLLMAgent(p, provider) for p in profiles}

# Build workflow
workflow = SequentialWorkflow(
    name="research-pipeline",
    agent_order=["researcher", "writer"],
)
for agent in agents.values():
    workflow.register_agent(agent)

# Seed state and run
state = State().add_message(
    Message(role=MessageRole.USER, content="Explain quantum computing in simple terms.")
)
final = asyncio.run(workflow.run(state))

for msg in final.messages:
    if msg.role == MessageRole.AI:
        print(f"[{msg.role.value}] {msg.content}\n")

---

Configuration

YAML Configuration

The canonical way to define agents is a YAML file with an agents list. Load it with AgentConfigLoader.

agents:
  # ── LiteLLM model string (preferred) ──────────────────────────────────
  - name: assistant          # Required. Unique. Alphanumeric + hyphens/underscores.
    description: "General-purpose assistant."   # Optional.
    system_prompt: "You are a helpful AI."      # Optional. Default: "You are a helpful AI assistant."
    model:
      litellm_model: "openai/gpt-4o"            # LiteLLM model string takes precedence.
      temperature: 0.7                           # 0.0–2.0  (default: 0.7)
      max_tokens: 4096                           # 1–1 000 000 (default: 4096)
      top_p: 1.0                                 # 0.0–1.0  (default: 1.0)
      top_k: null                                # ≥ 1, provider-dependent (default: null)
      extra_params:                              # Passed through to the provider.
        stop_sequences: ["DONE"]
    tools:
      - web_fetch
      - calculator
    max_iterations: 10        # 1–100 (default: 10)

  # ── Legacy provider + model_name pair ──────────────────────────────────
  - name: analyst
    system_prompt: "You are a precise data analyst."
    model:
      provider: anthropic              # "openai", "anthropic", "google", etc.
      model_name: claude-sonnet-4-6
      temperature: 0.3
      max_tokens: 8192
    tools: []

  # ── Extended thinking (Anthropic only) ────────────────────────────────
  - name: reasoner
    system_prompt: "Think step by step before answering."
    model:
      provider: anthropic
      model_name: claude-3-7-sonnet-20250219
      temperature: 1.0          # Required when extended_thinking is enabled.
      max_tokens: 16000
      extra_params:
        extended_thinking: true
        thinking_budget_tokens: 10000
    tools: []
    max_iterations: 5

Rules

• Every agent must specify either litellm_model or both provider + model_name.
• Agent names must be unique within a file.
• litellm_model takes precedence when both forms are present.

ModelConfig Reference

Field	Type	Default	Description
litellm_model	str | None	None	LiteLLM model string, e.g. "openai/gpt-4o"
provider	str | None	None	Provider name, e.g. "anthropic"
model_name	str | None	None	Model name, e.g. "claude-sonnet-4-6"
temperature	float	0.7	Sampling temperature (0.0–2.0)
max_tokens	int	4096	Max tokens to generate (1–1 000 000)
top_p	float	1.0	Nucleus sampling (0.0–1.0)
top_k	int | None	None	Top-k sampling (≥ 1, provider-dependent)
extra_params	dict	{}	Provider-specific pass-through parameters

ModelConfig.resolved_model returns a LiteLLM-compatible string:

• If litellm_model is set: returns it directly.
• Otherwise: returns "{provider}/{model_name}".

Programmatic Configuration

from miniAgent.core.config import AgentConfigLoader
from miniAgent.core.models.profile import AgentProfile, ModelConfig

# From YAML file
profiles = AgentConfigLoader.load_from_file("agents.yaml")

# From a Python dict (same structure as YAML)
profiles = AgentConfigLoader.load_from_dict({
    "agents": [
        {
            "name": "my-agent",
            "model": {"litellm_model": "openai/gpt-4o"},
            "system_prompt": "You are helpful.",
            "max_iterations": 5,
        }
    ]
})

# Directly via Pydantic model
profile = AgentProfile(
    name="coder",
    model=ModelConfig(
        provider="anthropic",
        model_name="claude-sonnet-4-6",
        temperature=0.2,
        max_tokens=8192,
    ),
    system_prompt="You are an expert Python programmer.",
    tools=["code_runner"],
    max_iterations=15,
)

---

Core Concepts

Concept	Class	Responsibility
Agent	BaseAgent	Wraps an LLM. Implements execute(), generate_response(), plan_tool_calls()
Profile	AgentProfile	Declarative config: name, system prompt, model, tools, max_iterations
State	State	Immutable runtime context: messages, variables, step count, status
Message	Message	One turn in a conversation: role + content + metadata
Tool	BaseTool	Async callable with JSON Schema exposed to the LLM
Workflow	BaseWorkflow	Orchestrates agents. Registers agents/tools then calls run()
Provider	BaseLLMProvider	Stateless LLM gateway: complete(), complete_with_tools(), stream()

State transitions are non-destructive — every State method (add_message, increment_step, set_variable) returns a new State instance via model_copy rather than mutating in place. The value objects Message, ToolCall, ToolResult, ModelConfig, AgentProfile, and ToolSchema are fully frozen Pydantic models.

---

Agents

LiteLLMAgent

LiteLLMAgent implements the full ReAct (Reason + Act) loop using any LiteLLM-supported model.

from miniAgent.adapters.llm.litellm_agent import LiteLLMAgent
from miniAgent.providers.litellm_provider import LiteLLMProvider
from miniAgent.core.models.profile import AgentProfile, ModelConfig

profile = AgentProfile(
    name="assistant",
    model=ModelConfig(litellm_model="openai/gpt-4o"),
    system_prompt="You are a helpful assistant.",
    max_iterations=10,
)

agent = LiteLLMAgent(profile, LiteLLMProvider())

# Execute against an existing State
final_state = await agent.execute(state)

# Or call lower-level methods directly
message = await agent.generate_response(messages)
tool_calls = await agent.plan_tool_calls(messages)

ReAct loop behaviour:

1. Builds message history from state, prepends system prompt if absent.
2. Calls complete_with_tools() if tools are registered, else complete().
3. Appends AI message to state; if tool calls present, invokes each tool.
4. Repeats until a text-only response or max_iterations is reached.

Constructor:

LiteLLMAgent(
    profile: AgentProfile,
    llm_provider: BaseLLMProvider,
    tools: dict[str, BaseTool] | None = None,
)

AnthropicAgent

AnthropicAgent inherits LiteLLMAgent but wires up AnthropicProvider internally. Use it when you need Claude-specific features (extended thinking, prompt caching) without going through LiteLLM.

from miniAgent.adapters.anthropic.agent import AnthropicAgent
from miniAgent.core.models.profile import AgentProfile, ModelConfig

profile = AgentProfile(
    name="reasoner",
    model=ModelConfig(
        provider="anthropic",
        model_name="claude-3-7-sonnet-20250219",
        temperature=1.0,
        max_tokens=16000,
        extra_params={"extended_thinking": True, "thinking_budget_tokens": 10000},
    ),
    system_prompt="Think step by step before answering.",
)

agent = AnthropicAgent(profile=profile, api_key="sk-ant-...")
final_state = await agent.execute(state)

Constructor:

AnthropicAgent(
    profile: AgentProfile,
    api_key: str | None = None,      # Falls back to ANTHROPIC_API_KEY env var
    tools: dict[str, BaseTool] | None = None,
)

GoogleADKAgent

GoogleADKAgent runs through Google's Agent Development Kit, using ADK's Runner and InMemorySessionService for session management. Like every other agent it implements BaseAgent, so it plugs into any workflow unchanged.

from miniAgent.adapters.google.agent import GoogleADKAgent
from miniAgent.core.models.profile import AgentProfile, ModelConfig

profile = AgentProfile(
    name="researcher",
    model=ModelConfig(model_name="gemini-2.5-flash"),
    system_prompt="You are a thorough research assistant.",
)

agent = GoogleADKAgent(profile)
final_state = await agent.execute(state)

Each execute() creates a fresh ADK session, runs the latest user message through the ADK Runner, and collects the final-response events into a single AI message.

Constructor:

GoogleADKAgent(profile: AgentProfile)   # model.model_name must be a Gemini model string

Requires: pip install "miniAgent[google]"

Building Custom Agents

Subclass BaseAgent and implement three methods:

from miniAgent.core.interfaces.agent import BaseAgent
from miniAgent.core.models.message import Message, ToolCall
from miniAgent.core.models.state import State

class MyCustomAgent(BaseAgent):
    async def execute(self, state: State) -> State:
        """Full ReAct loop — your main entry point."""
        # Call generate_response or plan_tool_calls as needed
        response = await self.generate_response(list(state.messages))
        return state.add_message(response).increment_step()

    async def generate_response(self, messages: list[Message]) -> Message:
        """Single-turn text generation (no tool handling)."""
        ...

    async def plan_tool_calls(self, messages: list[Message]) -> list[ToolCall]:
        """Determine which tools to invoke given the conversation."""
        ...

---

Workflows

All workflows share the same base interface:

workflow.register_agent(agent)          # Add an agent (raises ValueError on duplicate)
workflow.register_tool(tool)            # Add a tool (raises ValueError on duplicate)
final_state = await workflow.run(state) # Execute; returns final State

SequentialWorkflow

Runs agents one after another, passing state through the chain.

from miniAgent.orchestrator.sequential import SequentialWorkflow

workflow = SequentialWorkflow(
    name="pipeline",
    agent_order=["researcher", "writer", "editor"],  # Explicit order
)
workflow.register_agent(researcher_agent)
workflow.register_agent(writer_agent)
workflow.register_agent(editor_agent)

final = await workflow.run(initial_state)

• If agent_order is omitted, agents run in registration order.
• State flows unchanged between agents — each agent sees all prior messages.
• Raises WorkflowExecutionError if a named agent is not registered.

GraphWorkflow

Runs agents in a node-and-edge graph with conditional routing.

from miniAgent.orchestrator.graph import GraphWorkflow, END_NODE

wf = GraphWorkflow(name="review-loop", entry_point="writer")
wf.register_agent(writer_agent)
wf.register_agent(editor_agent)

# Deterministic edge: writer always goes to editor
wf.add_edge("writer", "editor")

# Conditional edge: editor routes based on state
def needs_revision(state: State) -> str:
    return "writer" if state.variables.get("needs_revision") else END_NODE

wf.add_conditional_edge("editor", needs_revision)

final = await wf.run()

Key rules:

• entry_point must be registered as an agent before calling run().
• END_NODE ("__END__") terminates the graph.
• Returning None from a router also terminates.
• If no edge is defined for a node, execution stops there.
• Exceeding state.max_steps raises WorkflowExecutionError.

Methods:

GraphWorkflow(name: str, entry_point: str | None = None)

wf.add_edge(source: str, target: str) -> None
# Deterministic. Raises ValueError if edge from source already exists.

wf.add_conditional_edge(source: str, router: Callable[[State], str | None]) -> None
# Dynamic. Raises ValueError if edge from source already exists.

ParallelWorkflow

Fans out to multiple agents concurrently (via asyncio.gather) then merges results.

from miniAgent.orchestrator.parallel import ParallelWorkflow

wf = ParallelWorkflow(name="research-fan-out")
wf.register_agent(web_search_agent)
wf.register_agent(arxiv_agent)
wf.register_agent(github_agent)

final = await wf.run(initial_state)
# final.messages contains all messages from all three agents, sorted by timestamp

Merge strategy:

• All new messages from every branch are collected and sorted by timestamp.
• Variables are merged with last-write-wins (ordered by agent_order).
• step_count is the sum of all branch step counts.
• Status is FAILED if any branch failed, otherwise COMPLETED.

LoopWorkflow

Runs agents repeatedly until an exit condition is satisfied or max_iterations is reached.

from miniAgent.orchestrator.loop import LoopWorkflow

def quality_sufficient(state: State) -> bool:
    return state.variables.get("quality_score", 0) >= 8

wf = LoopWorkflow(
    name="refinement-loop",
    agent_order=["refiner"],
    exit_condition=quality_sufficient,  # Optional; loop runs max_iterations if omitted
    max_iterations=5,
)
wf.register_agent(refiner_agent)

final = await wf.run(initial_state)

• Each iteration runs all agents in agent_order sequentially.
• Exit condition is evaluated after each full iteration.
• Without exit_condition, the loop always runs exactly max_iterations times.

WorkflowRegistry

A process-level store for named workflow instances, used by the REST API.

from miniAgent.orchestrator.registry import WorkflowRegistry

registry = WorkflowRegistry()
registry.register("my-pipeline", workflow)

wf = registry.get("my-pipeline")           # Raises WorkflowExecutionError if missing
registry.unregister("my-pipeline")
names = registry.list_workflows()           # -> ["my-pipeline", ...]
"my-pipeline" in registry                  # -> True / False
len(registry)                              # -> int

---

Tools

Tools extend agents with callable actions. miniAgent ships with 10 built-in tools and makes it trivial to add your own.

Built-in Tools

Import from miniAgent.tools — no extra installs needed (web tools require httpx, which is included in the default dependencies).

from miniAgent.tools import (
    CalculatorTool, TimestampTool, WordCountTool, TextTransformTool,
    JsonFormatterTool, UuidTool, EchoTool, FileReaderTool,
    WebFetchTool, WeatherTool,
)

Tool	Name	What it does	Requires
CalculatorTool	calculator	Safe arithmetic eval ((2+3)*4)	stdlib
TimestampTool	timestamp	Current UTC time (ISO 8601 or custom format)	stdlib
WordCountTool	word_count	Count words, characters, and lines	stdlib
TextTransformTool	text_transform	upper / lower / title / reverse / strip	stdlib
JsonFormatterTool	json_formatter	Parse and pretty-print JSON	stdlib
UuidTool	uuid_generator	Generate UUID v1 or v4	stdlib
EchoTool	echo	Echo input (useful for testing)	stdlib
FileReaderTool	file_reader	Read a local file (restricted to cwd)	stdlib
WebFetchTool	web_fetch	Fetch URL text content	httpx
WeatherTool	weather	Current weather via wttr.in (no API key)	httpx

Quick usage:

from miniAgent.tools import CalculatorTool, WeatherTool

agent = LiteLLMAgent(
    profile=profile,
    llm_provider=provider,
    tools={"calculator": CalculatorTool(), "weather": WeatherTool()},
)

Building Custom Tools

Subclass BaseTool, define a ToolSchema, and implement invoke():

from typing import Any
from miniAgent.core.interfaces.tool import BaseTool, ToolSchema

class WebSearchTool(BaseTool):
    def __init__(self) -> None:
        super().__init__(
            ToolSchema(
                name="web_search",
                description="Search the web for current information.",
                parameters={
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "Search query"},
                    },
                    "required": ["query"],
                },
            )
        )

    async def invoke(self, **kwargs: Any) -> str:
        query = kwargs["query"]
        # ... call your search API ...
        return f"Results for '{query}': ..."

Registering Tools with an Agent

tool = WebSearchTool()

# At agent construction time
agent = LiteLLMAgent(profile, provider, tools={"web_search": tool})

# Or on a workflow (shared across all agents)
workflow.register_tool(tool)

BaseTool interface

class BaseTool(ABC):
    schema: ToolSchema           # Exposed to the LLM via function calling

    @property
    def name(self) -> str:       # Returns schema.name

    @abstractmethod
    async def invoke(self, **kwargs: Any) -> str:
        """Execute the tool. Must return a string result."""

ToolSchema fields

Field	Type	Description
name	str	Unique tool name (used by LLM to call the tool)
description	str	What the tool does (LLM reads this)
parameters	dict	JSON Schema describing expected arguments

---

LLM Providers

LiteLLMProvider

Routes all calls through LiteLLM, supporting 80+ providers with a single interface.

from miniAgent.providers.litellm_provider import LiteLLMProvider
from miniAgent.core.models.profile import ModelConfig
from miniAgent.core.models.message import Message, MessageRole

provider = LiteLLMProvider()
model_cfg = ModelConfig(litellm_model="openai/gpt-4o")

messages = [Message(role=MessageRole.USER, content="Hello")]

# Non-streaming
response: Message = await provider.complete(messages, model_cfg)

# With tool calling
response, tool_calls = await provider.complete_with_tools(messages, model_cfg, tool_schemas)

# Streaming
async for chunk in provider.stream(messages, model_cfg):
    print(chunk, end="", flush=True)

Supported model string formats:

openai/gpt-4o
anthropic/claude-3-5-sonnet-20241022
google/gemini-1.5-pro
groq/llama-3.1-70b-versatile
bedrock/anthropic.claude-3-5-sonnet-20241022-v2:0
azure/my-deployment-name
vertex_ai/gemini-1.5-pro

API keys are read from provider-specific environment variables: OPENAI_API_KEY, ANTHROPIC_API_KEY, GEMINI_API_KEY, etc.

AnthropicProvider

Accesses Claude directly via the Anthropic SDK, unlocking features not available through LiteLLM.

from miniAgent.providers.anthropic_provider import AnthropicProvider
from miniAgent.core.models.profile import ModelConfig

provider = AnthropicProvider(api_key="sk-ant-...")   # or use ANTHROPIC_API_KEY
model_cfg = ModelConfig(
    provider="anthropic",
    model_name="claude-sonnet-4-6",
    temperature=0.7,
)

response = await provider.complete(messages, model_cfg)

Model strings are stripped of the anthropic/ prefix automatically, so both "claude-sonnet-4-6" and "anthropic/claude-sonnet-4-6" work.

Requires: pip install "miniAgent[anthropic]"

Extended Thinking

Extended thinking gives Claude extra computation time for complex reasoning. Enable it via extra_params:

from miniAgent.core.models.profile import ModelConfig

model_cfg = ModelConfig(
    provider="anthropic",
    model_name="claude-3-7-sonnet-20250219",
    temperature=1.0,          # Must be 1.0 when extended thinking is on
    max_tokens=16000,
    extra_params={
        "extended_thinking": True,
        "thinking_budget_tokens": 10000,   # Token budget for the thinking phase
    },
)

response = await provider.complete(messages, model_cfg)

# Thinking transcript is in response.metadata["thinking"]
print(response.metadata.get("thinking", ""))

Provider Registry

A lightweight singleton registry for the default provider:

from miniAgent.providers.registry import (
    get_default_provider,
    set_default_provider,
    reset_default_provider,
)

# Get (or lazily create) the default LiteLLMProvider
provider = get_default_provider()

# Swap it out (e.g. in tests or for a different backend)
set_default_provider(AnthropicProvider())

# Reset back to LiteLLMProvider
reset_default_provider()

---

State & Messages

State

Immutable runtime context passed through every workflow step.

from miniAgent.core.models.state import State, WorkflowStatus

state = State()                                # Fresh state
state = State(max_steps=500)                   # Custom step limit (default: 100)

# All mutation methods return a NEW State (State itself is not frozen, but
# these methods never mutate in place — they return a model_copy)
state = state.add_message(message)
state = state.increment_step()                 # Raises ValueError at max_steps
state = state.set_variable("key", value)

# Inspect
state.id              # str (UUID, auto-generated)
state.status          # WorkflowStatus enum
state.messages        # list[Message]
state.variables       # dict[str, Any]
state.step_count      # int
state.max_steps       # int
state.current_node    # str | None (set by workflows during execution)
state.created_at      # datetime (UTC, auto-generated)
state.updated_at      # datetime (UTC, refreshed on each transition)

WorkflowStatus values:

Value	Meaning
PENDING	Not yet started
RUNNING	Actively executing
COMPLETED	Finished successfully
FAILED	Terminated due to an error
CANCELLED	Explicitly cancelled

Message

One turn in a conversation thread.

from miniAgent.core.models.message import Message, MessageRole

msg = Message(
    role=MessageRole.USER,     # SYSTEM | USER | AI | TOOL
    content="Hello, world!",   # Required, min length 1
    metadata={"source": "api"},
)

msg.id            # str (UUID, auto-generated)
msg.timestamp     # datetime (UTC, auto-generated)
msg.role          # MessageRole
msg.content       # str
msg.metadata      # dict[str, Any]
msg.tool_call_id  # str | None (links a TOOL message to its originating ToolCall)

Message is a frozen Pydantic model — construct a new one rather than mutating an existing instance.

ToolCall and ToolResult

from miniAgent.core.models.message import ToolCall, ToolResult

# Created by the LLM via complete_with_tools()
call = ToolCall(
    tool_name="web_search",
    arguments={"query": "latest AI news"},
)

# Created by the tool invocation
result = ToolResult(
    tool_call_id=call.id,
    tool_name=call.tool_name,
    output="Here are the results...",
    is_error=False,
)

---

REST API

Running the Server

# Default: 0.0.0.0:8000
python -m miniAgent

# With environment overrides
HOST=127.0.0.1 PORT=9000 LOG_LEVEL=debug RELOAD=true python -m miniAgent

Environment variable	Default	Description
HOST	0.0.0.0	Bind address
PORT	8000	Bind port
LOG_LEVEL	info	Uvicorn log level
RELOAD	false	Enable auto-reload (development)
AGENTS_CONFIG_PATH	agents.yaml	Path to YAML agent config
CORS_ORIGINS	*	Comma-separated CORS origins

Interactive API docs are available at http://localhost:8000/docs.

Agent Endpoints

GET /agents

List all agents defined in the YAML configuration.

Response 200 OK:

[
  {
    "name": "researcher",
    "description": "Gathers and synthesizes information.",
    "provider": "openai",
    "model_name": "gpt-4o"
  }
]

Errors: 503 if configuration cannot be loaded.

---

GET /agents/{agent_name}

Get full configuration for a single agent.

Response 200 OK:

{
  "name": "researcher",
  "description": "Gathers and synthesizes information.",
  "provider": "openai",
  "model_name": "gpt-4o",
  "system_prompt": "You are a thorough research assistant.",
  "tools": ["web_search"],
  "max_iterations": 10
}

Errors: 404 if agent not found, 503 on config error.

---

POST /agents/{agent_name}/execute

Execute a single agent against a user message.

Request body:

{
  "message": "Explain quantum entanglement.",
  "variables": { "context": "for a general audience" }
}

Field	Type	Required	Description
message	string	Yes	User input (min 1 character)
variables	object	No	Initial state variables

Response 200 OK:

{
  "agent_name": "researcher",
  "response": "Quantum entanglement is...",
  "step_count": 3,
  "status": "completed"
}

Errors: 404 if agent not found, 500 on execution failure.

---

Workflow Endpoints

Workflows must be registered programmatically via the WorkflowRegistry before they can be called through the API.

from miniAgent.api.dependencies import get_workflow_registry

registry = get_workflow_registry()
registry.register("my-pipeline", my_workflow_instance)

GET /workflows

List all registered workflow names.

Response 200 OK:

[{ "name": "my-pipeline" }, { "name": "review-loop" }]

---

POST /workflows/{workflow_name}/run

Execute a workflow and wait for the final result.

Request body:

{
  "message": "Summarise the latest AI research.",
  "variables": { "format": "bullet_points" },
  "max_steps": 200
}

Field	Type	Required	Description
message	string | null	No	Optional seed message
variables	object	No	Initial state variables
max_steps	integer	No	Step limit (1–1000, default: 100)

Response 200 OK:

{
  "workflow_name": "my-pipeline",
  "status": "completed",
  "step_count": 8,
  "variables": { "quality_score": 9 },
  "message_count": 12
}

Errors: 404 if workflow not registered, 500 on execution failure.

---

POST /workflows/{workflow_name}/run/stream

Execute a workflow and stream progress as Server-Sent Events.

Request body: same as /run.

Response 200 OK — Content-Type: text/event-stream:

event: started
data: {"workflow": "my-pipeline"}

event: completed
data: {"workflow": "my-pipeline", "status": "completed", "step_count": 8, "message_count": 12, "variables": {}}

event: error
data: {"workflow": "my-pipeline", "detail": "Agent 'writer' failed: ..."}

Headers set automatically:

• Cache-Control: no-cache
• X-Accel-Buffering: no

Errors: 404 if workflow not registered (synchronous, before streaming starts).

---

Meta Endpoints

GET /health

{ "status": "ok", "version": "0.1.0" }

GET /

Redirects (307) to /docs.

---

Adapter Registry

The adapter registry maps provider names to BaseAgent subclasses, allowing create_agent(profile) to work without knowing the concrete type at call time.

from miniAgent.adapters.registry import (
    register_adapter,
    get_adapter,
    create_agent,
    list_providers,
    clear_registry,
)

# Register a custom adapter
register_adapter("my-provider", MyCustomAgent)

# Retrieve a factory
factory = get_adapter("my-provider")  # -> MyCustomAgent class

# Create an agent from a profile
agent = create_agent(profile)          # profile.model.provider must be registered
                                       # Raises AdapterLoadError if not

# Introspect
list_providers()                       # -> ["anthropic", "my-provider", ...]
clear_registry()                       # Remove all (useful in tests)

Notes:

• Provider names are case-insensitive.
• The AnthropicAgent adapter is auto-registered at import time if the anthropic SDK is installed.
• create_agent() requires profile.model.provider to be set (not just litellm_model).

---

Exception Hierarchy

All exceptions inherit from MiniAgentError. Catch specific types to handle different failure modes.

MiniAgentError(message, cause=None)
├── ConfigurationError        YAML invalid, file not found, validation failure
├── AdapterLoadError          Provider not registered or factory failed
├── AgentExecutionError       LLM or tool error inside agent.execute()
├── ToolInvocationError       Tool.invoke() raised an error
├── WorkflowExecutionError    Workflow-level failure (bad config, max steps, etc.)
├── StateValidationError      State mutation violated constraints
├── LLMProviderError          Provider API call failed (network, auth, quota)
└── FrameworkEngineError      External framework SDK (Google ADK, etc.) failed

from miniAgent.core.exceptions import (
    MiniAgentError,
    ConfigurationError,
    AgentExecutionError,
    WorkflowExecutionError,
    LLMProviderError,
    ToolInvocationError,
)

try:
    final = await workflow.run(state)
except WorkflowExecutionError as e:
    print(f"Workflow failed: {e}")
    if e.__cause__:
        print(f"Caused by: {e.__cause__}")
except LLMProviderError as e:
    print(f"LLM call failed: {e}")
except MiniAgentError as e:
    print(f"Framework error: {e}")

---

Architecture

miniAgent/
├── core/                        # Framework abstractions (no vendor dependencies)
│   ├── interfaces/
│   │   ├── agent.py             #   BaseAgent (execute, generate_response, plan_tool_calls)
│   │   ├── workflow.py          #   BaseWorkflow (register_agent, register_tool, run)
│   │   ├── tool.py              #   BaseTool (invoke), ToolSchema
│   │   └── provider.py          #   BaseLLMProvider (complete, complete_with_tools, stream)
│   ├── models/
│   │   ├── message.py           #   Message, ToolCall, ToolResult, MessageRole
│   │   ├── profile.py           #   AgentProfile, ModelConfig
│   │   ├── state.py             #   State, WorkflowStatus
│   │   └── workflow_config.py   #   WorkflowConfig
│   ├── config.py                #   AgentConfigLoader (YAML → AgentProfile[])
│   └── exceptions.py            #   Typed exception hierarchy
│
├── providers/                   # LLM provider implementations
│   ├── litellm_provider.py      #   LiteLLMProvider (80+ backends via LiteLLM)
│   ├── anthropic_provider.py    #   AnthropicProvider (native SDK, extended thinking)
│   └── registry.py             #   get/set/reset default provider singleton
│
├── adapters/                    # Agent factory adapters
│   ├── registry.py              #   register_adapter, create_agent, get_adapter
│   ├── llm/
│   │   └── litellm_agent.py     #   LiteLLMAgent (ReAct loop)
│   ├── anthropic/
│   │   └── agent.py             #   AnthropicAgent (native Claude)
│   └── google/
│       └── agent.py             #   GoogleADKAgent (pip install miniAgent[google])
│
├── orchestrator/                # Workflow execution engines
│   ├── sequential.py            #   SequentialWorkflow
│   ├── parallel.py              #   ParallelWorkflow (asyncio.gather fan-out)
│   ├── loop.py                  #   LoopWorkflow (exit condition or max_iterations)
│   ├── graph.py                 #   GraphWorkflow (conditional routing, END_NODE)
│   ├── registry.py              #   WorkflowRegistry
│   └── memory/
│       └── context.py           #   ExecutionContext (conversation memory)
│
├── tools/                       # Built-in tool implementations
│   ├── calculator.py            #   CalculatorTool (safe arithmetic)
│   ├── datetime_tools.py        #   TimestampTool
│   ├── text_tools.py            #   WordCountTool, TextTransformTool, JsonFormatterTool
│   ├── system_tools.py          #   UuidTool, EchoTool, FileReaderTool
│   └── web_tools.py             #   WebFetchTool, WeatherTool (httpx, no API key)
│
└── api/                         # FastAPI REST layer
    ├── app.py                   #   create_app() factory (CORS, logging middleware)
    ├── dependencies.py          #   get_agent_profiles, get_agent_by_name, get_workflow_registry
    └── routes/
        ├── agent_routes.py      #   GET /agents, GET /agents/{name}, POST /agents/{name}/execute
        ├── workflow_routes.py   #   GET /workflows, POST /workflows/{name}/run[/stream]
        └── meta_routes.py       #   GET /health, GET / (redirect)

Data flow for a workflow execution:

Client → POST /workflows/{name}/run
       → get_workflow_registry().get(name)
       → workflow.run(seeded_state)
         → for each agent: agent.execute(state)
           → provider.complete_with_tools(messages, model_cfg, tools)
             → litellm.acompletion(...) / anthropic.messages.create(...)
           → tool.invoke(**args)  [if tool calls present]
           → state = state.add_message(...).increment_step()
       → WorkflowRunResponse(status, step_count, variables, message_count)

---

Examples

The examples/ directory contains runnable scripts showing each core concept:

Script	What it demonstrates
01_simple_agent.py	Single agent, no tools
02_tool_agent.py	ReAct loop with Calculator, Timestamp, Weather
03_pipeline.py	SequentialWorkflow: researcher → writer
04_graph_workflow.py	GraphWorkflow with conditional routing

Run any example:

export OPENAI_API_KEY="sk-..."
python examples/01_simple_agent.py

Advanced adapter examples (CrewAI, LangChain) are in examples/advanced/.

---

Testing

# Run all tests
pytest

# With coverage report
pytest --cov=miniAgent --cov-report=term-missing

# Run only unit tests
pytest tests/unit/ -v

# Run only integration tests
pytest tests/integration/ -v

The test suite requires no live API keys — all LLM calls are mocked.

tests/
├── unit/                    # Per-module unit tests (mocked LLM)
│   ├── test_models.py
│   ├── test_litellm_provider.py
│   ├── test_anthropic_provider.py
│   ├── test_litellm_agent.py
│   ├── test_anthropic_agent.py
│   ├── test_orchestrators.py
│   ├── test_parallel_loop_workflows.py
│   ├── test_registry.py
│   ├── test_config.py
│   ├── test_api.py
│   └── test_api_routes.py
└── integration/             # Cross-component integration tests
    ├── conftest.py          #   Shared fixtures (mock_litellm, base_state, ...)
    ├── test_sequential_workflow.py
    ├── test_graph_workflow.py
    ├── test_parallel_workflow.py
    ├── test_loop_workflow.py
    ├── test_tool_use.py
    ├── test_api.py
    └── test_config_roundtrip.py

---

Development

# Install with all dev dependencies
pip install -e ".[dev]"

# Type checking (strict mypy)
mypy miniAgent/

# Linting and formatting
ruff check miniAgent/
ruff check --fix miniAgent/

# Full check (lint + types + tests + coverage)
ruff check miniAgent/ && mypy miniAgent/ && pytest --cov=miniAgent --cov-fail-under=100

---

License

MIT — see LICENSE.