mini-agent-framework 0.8.7

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
Parallel Chat Research	chat() auto-detects heavy multi-part requests (read 10 files, search many topics) and spawns parallel read-only research agents, then merges findings
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
Run Mode = Actions Only	run() never direct-answers — every task becomes a plan executed by sub-agents (use chat() for discussion)
Plan-First Architect	Before work starts, an architect agent writes PLAN.md + todo.json (checklist) into the project workdir
Checklist Completion	Every written file is marked done in todo.json; a completion pass keeps building until the whole project is complete
Professional Workdir	PLAN.md/todo.json live in a platform directory (%APPDATA%\mini_agent\projects, ~/.local/share/mini_agent/projects) — never in CWD; override with workdir= or MINI_AGENT_WORKDIR
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, projects, 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.run()
    │
    ├─ Planner (LLM)            ← plans the task; run mode NEVER direct-answers
    │     │
    │     ▼
    ├─ Architect                ← writes PLAN.md + todo.json (checklist) to workdir
    │     │
    │     ▼
    └─ Need-based execution     ← agents are spawned to match the plan
        ├─ Level 0: Agent A ───┐
        ├─ Level 1: Agent B ───┤ (parallel, dependency-aware)
        ├─ Level 2: Agent C ───┘
        │     each agent marks written files done in todo.json
        ▼
    └─ Completion pass         ← re-reads todo.json, finishes pending files,
        │                        outputs the FINAL ANSWER (start-to-end summary)
        ▼
    └─ Verification loop       ← run_verification tool: build → check → fix → repeat

The Orchestrator is the central controller. It:

1. Plans the task using an LLM — in run mode it always produces a plan (no direct answers)
2. Architects the full file list into PLAN.md + todo.json in the workdir
3. Spawns agents on demand, each with the right tools and skills
4. Executes agents in dependency-respecting parallel levels, tracking the checklist
5. Completes — a completion pass keeps going until every planned file exists, then returns the final summary

# Where PLAN.md + todo.json are saved (never CWD):
#   Windows:  %APPDATA%\mini_agent\projects\
#   macOS:    ~/Library/Application Support/mini_agent/projects/
#   Linux:    ~/.local/share/mini_agent/projects/

# Override per-run:
orch = Orchestrator(llm, workdir="/content/drive/MyDrive/my_project")
# or via env var:
#   export MINI_AGENT_WORKDIR=/path/to/projects

---

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)	Auto — parallel researchers for heavy multi-part reads/searches
run()	Complex task execution	Full access	Yes — always plans + sub-agents, never direct-answers

Parallel Research in Chat

When a chat message asks to read many files or search many independent topics at once (e.g. "read these 10 files and summarize", "compare 5 libraries"), a lightweight coordinator LLM call detects it and splits the work:

chat("read 10 files and summarize")
   ↓
1. Coordinator: {"needs_parallel_research": true, "research_tasks": [files 1-5, files 6-10]}
   ↓
2. N parallel read-only research agents (ThreadPoolExecutor)
   ↓
3. One synthesis call merges findings → final answer

• Simple questions skip this entirely — single agent as before (one extra cheap coordinator call)
• Research agents only get read-only tools (read_text_file, web_search, grep, ...) — nothing can be written
• max_agents_auto applies here too (None = unlimited)

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.

Plan-First Checklist (PLAN.md + todo.json)

Every run() task starts with an architect agent that writes two files into the workdir:

File	Contents
PLAN.md	Full project plan — architecture, file list, data flow, setup instructions
todo.json	Checklist — {"path": "status"} per planned file (pending → in_progress → done)

Then:

1. Worker agents build the files and mark them done in todo.json
2. A completion pass re-reads the checklist and continues building any [Not fully complete] pending files
3. Completion rounds loop (default 3, max_completion_rounds=) until the whole project exists
4. The completion agent's output (start-to-end summary) is the final answer

The checklist lives in the same workdir as the project files — agents read/write todo.json with absolute paths, so state survives even long builds.

---

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"])

Simple Task (still action-oriented)

Even simple run() tasks go through a worker agent — run mode never returns a direct answer:

result = orch.run("What is 2+2?")
print(result["final_answer"])  # worker agent computes and returns the answer

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_AGENTS_AUTO	None	Auto need-based cap (None = unlimited — agents spawn purely by need; set a number to guard runaway plans)
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	10	LTM summarization frequency
DEFAULT_WORKDIR	(platform dir)	Where PLAN.md + todo.json are written (env: MINI_AGENT_WORKDIR)
PLAN_FILENAME	PLAN.md	Plan artifact filename
TODO_FILENAME	todo.json	Checklist artifact filename

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