mini-agent-framework 0.8.3

A dynamic, need-based multi-agent AI framework with session memory, action tracking, tool approval, and browser automation

mini-agent-framework

A dynamic, need-based multi-agent AI framework for Python.
No fixed agent pool — the Orchestrator plans the task, spawns worker agents with matched tools and skills at runtime, and aggregates their results into one coherent answer.

---

Features

Feature	Description
Dynamic Agent Spawning	No pre-defined pool — agents are created on-the-fly per task
Need-Based Planning	LLM-powered planner decomposes tasks into sub-tasks automatically
Parallel Execution	Sub-tasks run concurrently with dependency-aware scheduling
Tool System	Built-in file (read, write, edit, glob, grep), web, math, data, system, bash, and browser tools
Skill System	Domain expertise via Markdown files injected into agent prompts
Session Management	Multi-turn chat with persistent, named sessions
Action Tracking	Real-time event logging (plan, agent lifecycle, tool calls, aggregation)
Tool Approval	Gated execution with optional user approval callback
Chat with Tools	chat() uses read-only tools for research before answering
Custom Providers	Extend BaseLLMProvider for any LLM (OpenAI, Anthropic, Ollama, etc.)
Custom Tools	Wrap any Python function as a tool with automatic parameter detection
Circuit Breaker	Automatic failure detection — stops cascading failures after N consecutive errors, then recovers
Dead Letter Queue	Failed tasks are queued with error context for later inspection or retry
Task Persistence	Every task's state is saved to disk — survives crashes and restarts
Inter-Agent Messaging	Agents communicate via a message bus (send, broadcast, handler pattern)
Session Cleanup	Idle and absolute timeout policies with background thread auto-cleanup
Structured Logging	JSON-formatted logs with timestamps, events, and structured fields
Connection Pooling	Shared HTTPX client across provider instances — reduces connection overhead
Async Tool Execution	execute_async() for non-blocking tool calls with async retry/timeout
Idempotency Keys	Prevents duplicate planning/execution of the same task
Auto Memory Pruning	Importance-scored turns automatically pruned when token budget is exceeded
Structured Memory	Facts and topics stored with categories, verification status, and mention tracking
Plan Approval	Optional human-in-the-loop callback to approve or reject plans before execution
Agent Handoff	An agent can request handoff via "HANDOFF" in its response — a new agent continues
Dynamic Re-planning	If a sub-agent fails, the orchestrator automatically re-plans and retries
Verification Loop Tool	run_verification tool — agents call it after building; auto-detects language, runs syntax/lint/tests/logic checks, and fixes failures in a loop. Auto-registered.
Platform-Aware Paths	Sessions, task store, and browser data auto-detect Windows/macOS/Linux paths

---

Installation

pip install mini-agent-framework

With browser automation

pip install mini-agent-framework[browser]

---

Quick Start

import os
from mini_agent import Orchestrator, NvidiaProvider
from mini_agent.registry.builtin import FILE_TOOLS, WEB_TOOLS

llm = NvidiaProvider(api_key=os.environ["NVIDIA_API_KEY"])
orch = Orchestrator(llm)
orch.register_tools(FILE_TOOLS + WEB_TOOLS)

result = orch.run("Search for AI news and save to ai_news.txt")
print(result["final_answer"])

---

Core Architecture

User Task
    │
    ▼
Orchestrator._plan()          ← LLM plans: single-agent or multi-agent?
    │
    ├─ Single-agent (simple task)
    │   └─ Agent.run() → tool_loop → final answer
    │
    └─ Multi-agent (complex task)
        ├─ Level 0: Agent A ───┐
        ├─ Level 1: Agent B ───┤ (parallel, dependency-aware)
        ├─ Level 2: Agent C ───┘
        └─ Orchestrator._aggregate() → merged final answer

The Orchestrator is the central controller. It:

1. Plans the task using an LLM — decides if sub-agents are needed
2. Spawns agents on demand, each with the right tools and skills
3. Executes agents in dependency-respecting parallel levels
4. Aggregates all agent results into one final answer

---

LLM Providers

Built-in: NvidiaProvider

Uses NVIDIA NIM (OpenAI-compatible endpoint):

from mini_agent import NvidiaProvider

# Uses NVIDIA_API_KEY from environment
llm = NvidiaProvider()

# Or explicit configuration
llm = NvidiaProvider(
    api_key="nvapi-...",
    model="deepseek-ai/deepseek-v4-flash",
    temperature=0.7,
    top_p=0.95,
    max_tokens=4096,
)

Default model: deepseek-ai/deepseek-v4-flash.
API key must start with nvapi-. Get one at build.nvidia.com.
NVIDIA free tier is ~40 requests/minute — the provider retries 5 times with exponential backoff on 429 errors.

Custom Provider

Implement BaseLLMProvider for any LLM (OpenAI, Anthropic, Ollama, Groq, etc.):

from mini_agent import BaseLLMProvider

class MyProvider(BaseLLMProvider):
    def generate_stream(self, system_prompt: str, user_message: str):
        # Must be a generator yielding tokens one by one
        for token in my_api_call(system_prompt, user_message):
            yield token

generate() is implemented automatically by accumulating tokens from generate_stream().

---

Tools

Tool Class

Every tool is a Tool dataclass wrapping a Python function:

from mini_agent import Tool

def get_weather(city: str) -> str:
    return f"{city}: 32°C sunny"

tool = Tool(
    name="get_weather",
    description="Get current weather for a city",
    func=get_weather,
    parameters={"city": "str"},    # auto-detected if omitted
    requires_approval=False,        # gates execution behind approval callback
    read_only=True,                 # available during planning phase
    validator=None,                 # input validation function
    on_error=None,                  # error handler (returns fallback string)
)

Built-in Tool Categories

from mini_agent.registry.builtin import (
    BUILTIN_TOOLS,   # FILE + WEB + DATA + MATH + SYSTEM + BASH
    FILE_TOOLS,      # read_file, write_file, edit_file, glob, grep, ...
    WEB_TOOLS,       # web_search, web_fetch, ...
    DATA_TOOLS,      # read_json, write_json
    MATH_TOOLS,      # calculator, uuid, random, word_count
    SYSTEM_TOOLS,    # datetime, cwd
    BASH_TOOL,       # bash (with workdir & timeout)
    BROWSER_TOOLS,   # Playwright browser automation (opt-in)
    ALL_TOOLS,       # BUILTIN_TOOLS + BROWSER_TOOLS
)

Tool Approval

Tools with requires_approval=True will prompt the user before execution:

from mini_agent import Orchestrator
from mini_agent.core.utils import cli_approval_callback

orch = Orchestrator(
    llm,
    approval_callback=cli_approval_callback,
    plan_approval_callback=None,              # Human-in-the-loop plan approval
    task_store_dir="./task_store",            # Task persistence directory
    dlq_file="./dead_letter_queue.json",      # Dead letter queue file
    circuit_breaker_threshold=5,              # Failures before circuit opens
)

Built-in callbacks:

Callback	Behavior
cli_approval_callback	Terminal prompt [y/N]
auto_approve_callback	Always approves (logging mode)
auto_reject_callback	Always rejects (dry-run / read-only)

Custom callbacks follow the signature (tool_name: str, arguments: dict) -> bool.

Registering Tools

# Single
orch.register_tool(tool)

# Multiple
orch.register_tools([tool1, tool2])

# Built-in categories
orch.register_tools(FILE_TOOLS + WEB_TOOLS + MATH_TOOLS)

---

Skills

Skills provide domain expertise as Markdown files. They are matched against the task and injected into agent prompts.

Bundled Skills

Skills are pre-defined Skill objects you can import and register just like tools:

from mini_agent.skills.builtin import SKILLS, CODE_REVIEW, DEVELOPER

# Register all bundled skills
orch.register_skills(SKILLS)

# Or select individual skills
orch.register_skills(CODE_REVIEW + DEVELOPER)

Custom Skills

Skills provide domain expertise as Markdown files. They are matched against the task and injected into agent prompts:

---
name: code-review
description: Review Python code for bugs, style issues, and security vulnerabilities
---

Your detailed expertise and instructions go here...

from mini_agent import Skill

# Single skill
orch.register_skill(Skill(
    name="code-review",
    description="Review Python code for bugs, style, security",
))

# Load from a directory of .md skill files
orch.load_skills_from_dir("path/to/skills/")

# Multiple
orch.register_skills([skill1, skill2])

The planner LLM decides which skill (if any) a worker needs, via the skill field in the plan JSON — same way it selects tools. Only the planned skill is injected into the agent prompt.

---

Session Management

Manage multi-turn conversations with persistent storage:

from mini_agent import Orchestrator, SessionManager, NvidiaProvider

sm = SessionManager()
orch = Orchestrator(llm, session_manager=sm)

# Create a session
s1 = sm.create_session("AI Discussion")

# Multi-turn chat (streaming, context-aware)
orch.chat("What is machine learning?", session_id=s1["id"])
orch.chat("Explain neural networks", session_id=s1["id"])

# List all sessions
print(sm.list_sessions())

Session API

Method	Description
create_session(name)	Create new session
list_sessions()	List all sessions with metadata
get_session(id)	Get session memory
delete_session(id)	Delete a session
rename_session(id, name)	Rename a session

Chat vs Run

Method	Use Case	Web/File Tools	Agents
chat()	Conversational Q&A + research	Read-only (research)	No
run()	Complex task execution	Full access	Yes (multi-agent)

Verification Loop Tool

The verification loop is available as a tool (run_verification) that auto-registers when the orchestrator is created. Agents can call it after implementing a task:

orch = Orchestrator(llm)
orch.register_tools(BUILTIN_TOOLS)
orch.register_skills(DEVELOPER)

# Agent builds first, then calls run_verification tool on its own
result = orch.run("build a REST API with Flask")

The loop works for any language or framework:

Phase	What happens
Build	Agent writes code as requested
Verify	AI discovers language/framework, runs syntax check, lint, tests, and logic review
Fix	If issues found, agent reads the code, understands bugs, and fixes them
Repeat	Loop continues until all checks pass or max iterations (default 5) reached
Ask	At loop limit, asks user: continue fixing or accept as-is?

The tool requires user approval (requires_approval=True), so the agent will ask before running the verification loop. Pass approval_callback to Orchestrator() to control this behavior.

---

Action Tracking

Real-time event monitoring for debugging and observability:

from mini_agent import Orchestrator, ActionTracker, console_event_logger

tracker = ActionTracker(on_event=console_event_logger)
orch = Orchestrator(llm, action_tracker=tracker)

The default console_event_logger prints formatted events to stderr with color support:

╔══ PLAN ═══════════════════════════════════════
║  multi-agent  (2 sub-tasks)
║  #0  researcher  [web_search]
║  #1  coder  [write_text_file, bash]  ← after #0
╚═══════════════════════════════════════════════

    ┌══ WORKER: researcher ═══════════════════════┐
    │  Search for latest AI frameworks
    │  Tools: web_search, fetch_url
    └══════════════════════════════════════════════┘
    │  1/∞  ⚙ web_search(q=AI trends 2026)
    │        → Found 3 major frameworks...
    │  ✔ Research complete

    ┌══ WORKER: coder ═══════════════════════════┐
    │  Write code based on research
    │  Tools: write_text_file, bash
    │  Skills: developer
    └══════════════════════════════════════════════┘
    │  1/5  ⚙ write_text_file(path=output.py, ...)
    │        → Wrote 500 characters to output.py

╔══ AGGREGATOR ═══════════════════════════════════
║  Merging agent outputs...
╚══════════════════════════════════════════════════

Events

Event	Triggered when
plan	Orchestrator creates a task plan
agent_start	A worker agent is spawned
agent_end	A worker agent completes
tool_call	An agent calls a tool
tool_result	A tool returns its result
aggregate	Orchestrator merges agent results
token	A streaming token is produced
research	A research action occurs

Custom event handler:

def my_handler(event_type: str, data: dict):
    print(f"[{event_type}] {data}")

tracker = ActionTracker(on_event=my_handler)

---

Examples

Multi-Agent Task

result = orch.run(
    "Search for top 3 Python web frameworks, "
    "compare their features, and save the comparison to comparison.md"
)
print(result["final_answer"])

Single Direct Task

Simple tasks bypass the multi-agent system:

result = orch.run("What is 2+2?")
print(result["final_answer"])  # "4"

Interactive CLI

from mini_agent import (
    Orchestrator, NvidiaProvider, SessionManager, Tool, Skill,
    ActionTracker, console_event_logger
)
from mini_agent.registry.builtin import FILE_TOOLS, WEB_TOOLS, MATH_TOOLS
from mini_agent.skills.builtin import CODE_REVIEW

llm = NvidiaProvider(api_key=os.environ["NVIDIA_API_KEY"])
sm = SessionManager()
tracker = ActionTracker(on_event=console_event_logger)
orch = Orchestrator(llm, session_manager=sm, action_tracker=tracker)
orch.register_tools(FILE_TOOLS + WEB_TOOLS + MATH_TOOLS)
orch.register_skills(CODE_REVIEW)

session = sm.create_session("Interactive")

while True:
    user_input = input("\n> ").strip()
    if user_input.lower() in ("exit", "quit"):
        break
    if user_input.startswith("!"):
        result = orch.run(user_input[1:], session_id=session["id"])
        print(f"\n{result['final_answer']}")
    else:
        orch.chat(user_input, session_id=session["id"])

---

Browser Tools (Opt-in)

Requires undetected-chromedriver (auto-installed with [browser] extra):

pip install mini-agent-framework[browser]

from mini_agent.registry.builtin import init_browser, BROWSER_TOOLS

# headless=True  → invisible (default)
# headless=False → visible GUI window
init_browser(headless=False)
orch.register_tools(BROWSER_TOOLS)

Available: browser_open, browser_click, browser_fill, browser_select, browser_scroll, browser_extract, browser_screenshot, browser_read, browser_observe, browser_navigate, browser_tabs, browser_download, browser_dialog, browser_wait, browser_check, browser_close, browser_javascript, browser_upload.

To toggle between modes at runtime:

init_browser(headless=False)  # switch to visible mode
init_browser(headless=True)   # switch back to headless

---

Circuit Breaker

Prevents cascading failures. After N consecutive tool failures, the circuit opens — all further calls fail fast with CircuitBreakerOpenError. After a recovery timeout, it transitions to half-open, testing one call. Success closes the circuit; failure re-opens it.

from mini_agent.core.circuit_breaker import CircuitBreaker, CircuitState

cb = CircuitBreaker(failure_threshold=3, recovery_timeout=30.0)
print(cb.state)  # CircuitState.CLOSED

# Orchestrator uses it automatically for all tool executions
result = orch.execute_tool("calculator", {"expression": "2+2"})

---

Dead Letter Queue

Failed tasks are automatically pushed to a DLQ with error context. Inspect or retry them later.

from mini_agent.core.dead_letter_queue import DeadLetterQueue

dlq = DeadLetterQueue(persist_file="./failures.json")
print(len(dlq))           # Number of failed tasks
entry = dlq.pop()          # Oldest failed task
successes, failures = dlq.retry_all(executor_fn, max_retries=3)

# Access via Orchestrator
orch = Orchestrator(llm, dlq_file="./dead_letter_queue.json")
print(len(orch.dead_letter_queue))

---

Task Persistence

Every task node's state (pending, running, completed, failed) is saved as a JSON file. Survives restarts.

# Automatic via Orchestrator
orch = Orchestrator(llm, task_store_dir="./task_store")

# Manual use
from mini_agent.core.task_persistence import TaskPersistenceStore
from mini_agent.core.graph import TaskNode

store = TaskPersistenceStore(storage_dir="./task_store")
node = TaskNode(id="task_1", description="Research", dependencies=[])
node.mark_completed("Done")
store.save_task(node, "run_abc")
loaded = store.load_task("run_abc", "task_1")
store.list_run_ids()      # ["run_abc", ...]
store.list_tasks("run_abc")  # ["task_1"]
store.purge_run("run_abc")

---

Inter-Agent Communication

Agents can send and receive messages via a MessageBus. Supports point-to-point send, broadcast, priority levels, and handler registration.

from mini_agent.core.inter_agent import MessageBus, MessagePriority

bus = MessageBus()
mailbox_a = bus.register_agent("agent_a")
mailbox_b = bus.register_agent("agent_b")

bus.send("agent_a", "agent_b", "greeting", {"text": "hello"})
bus.broadcast("agent_a", "announce", {"msg": "update"})

# Register a handler
mailbox_b.register_handler("greeting", lambda msg: print(msg.payload))

# Poll for messages
unread = mailbox_b.poll("greeting")

Available via Orchestrator:

orch.send_agent_message("agent_a", "agent_b", "request_data", {"query": "sales"})
orch.broadcast_agent_message("coordinator", "status_update", {"status": "done"})

---

Session Cleanup

Automatic cleanup of idle or expired sessions via a background thread.

from mini_agent.core.session_cleanup import SessionCleanupManager, SessionTimeoutPolicy

policy = SessionTimeoutPolicy(idle_timeout=1800, absolute_timeout=86400)
manager = SessionCleanupManager(policy=policy, on_cleanup=lambda sid: print(f"{sid} expired"))
manager.register("session_1")
manager.start(interval=60.0)   # Check every 60 seconds
manager.stop()

Integrated into SessionManager:

sm = SessionManager(idle_timeout=1800, absolute_timeout=86400)
# Cleanup runs automatically in the background

---

Structured Logging

All framework events are logged as JSON objects with consistent structure.

from mini_agent.core.logging import StructuredLog, get_logger

log = StructuredLog(name="my_agent", level="INFO")
log.info("user_query", query="What is AI?", user_id=42)
log.error("task_failed", exc_info=exception, task_id="t1")

# Singleton — same instance across all modules
logger = get_logger()

Output:

{"timestamp": "2026-07-24T12:00:00", "level": "INFO", "event": "user_query", "query": "What is AI?", "user_id": 42}

---

Async Tool Execution

Non-blocking tool execution with async retry and timeout:

import asyncio
from mini_agent import ToolExecutor, Tool

executor = ToolExecutor()

async def main():
    result = await executor.execute_async(
        tool, {"param": "value"}, task_id="t1", agent_id="a1"
    )
    print(result)

asyncio.run(main())

---

Connection Pooling

NvidiaProvider shares a single HTTPX client across all instances via reference counting. The first provider creates the client; subsequent instances reuse it. Call close() to decrement the refcount.

p1 = NvidiaProvider(api_key="nvapi-xxx")
p2 = NvidiaProvider(api_key="nvapi-xxx")
print(p1.client is p2.client)  # True — same shared client

p1.close()  # refcount -= 1
p2.close()  # refcount == 0 → client closed

---

Configuration

Settings in mini_agent.config.settings:

Setting	Default	Description
MAX_AGENTS	5	Maximum parallel sub-agents per task
MAX_RECURSION_DEPTH	2	Maximum nested agent spawn depth
MAX_TOOL_ITERATIONS	0	Maximum tool calls per agent (0 = unlimited)
MEMORY_MAX_TURNS	5	Conversation turns retained (non-session)
SESSION_MAX_TURNS	0	Turns stored per session (0 = unlimited)
MEMORY_CONTEXT_TURNS	2	Recent turns included in agent prompt
MAX_CONTEXT_TOKENS	4000	Token budget for compressed context
SUMMARIZE_EVERY_N_TURNS	5	LTM summarization frequency

from mini_agent.config.settings import MAX_AGENTS, MAX_CONTEXT_TOKENS
MAX_AGENTS = 10
MAX_CONTEXT_TOKENS = 8000

---

Dependencies

Package	Version
openai	>=1.0
requests	>=2.31
pydantic	>=2.8
python-dotenv	>=1.0
aiohttp	>=3.9
fastapi	>=0.111
uvicorn	>=0.24

Optional: playwright>=1.38 (browser tools)

---

Development

# Clone
git clone https://github.com/ayyandurai111/mini-agent-framework.git
cd mini-agent-framework

# Install in editable mode
pip install -e .

# With browser support
pip install -e .[browser]

---

License

MIT License — see LICENSE.

---

Links

• PyPI: mini-agent-framework
• Repository: github.com/ayyandurai111/mini-agent-framework
• Issues: github.com/ayyandurai111/mini-agent-framework/issues