Content
# mini-crew
A lightweight multi-agent framework — **CrewAI-style role orchestration + event-driven Flow engine**, pure Python. Define agents, give them tools, and run them in crews or event-driven pipelines. Supports MCP for tooling.
## Quick start
```bash
pip install -e .
```
Set your API key in `.env`:
```
DEEPSEEK_API_KEY=your-key-here
DEEPSEEK_MODEL=deepseek-v4-pro
```
### 5-minute example
```python
from mini_crew import Agent, Task, Crew, Process, tool
@tool
def calculator(expression: str) -> str:
"""Safely evaluate a math expression."""
import math
safe = {"abs": abs, "sqrt": math.sqrt, "pi": math.pi}
return str(eval(expression, {"__builtins__": {}}, safe))
agent = Agent(
name="Math Helper",
role="Math Expert",
goal="Solve calculations accurately",
backstory="You use the calculator tool for precise math.",
tools=[calculator],
verbose=True,
)
task = Task(
description="Calculate (123 + 456) * 789",
expected_output="The numeric result",
agent=agent,
)
crew = Crew(agents=[agent], tasks=[task], process=Process.sequential)
result = crew.kickoff()
print(result.results[0]["output"])
```
## What you get
Four building blocks. That's it.
**Agent** — an LLM-powered worker with a role, goal, backstory, and a set of tools. It calls the LLM, uses tools via function calling, and produces results.
```python
agent = Agent(
name="Researcher",
role="Information Gatherer",
goal="Collect and summarize information",
backstory="Experienced researcher, thorough and detail-oriented.",
llm="deepseek-v4-pro",
tools=[web_search, calculator],
verbose=True,
)
result = agent.execute("What's the GDP of France?")
```
**Task** — a unit of work tied to an agent. Tasks can depend on other tasks via `context`, and you can mark them `async_execution=True` to run in parallel with siblings.
```python
task = Task(
description="Analyze the quarterly report",
expected_output="JSON summary with key metrics",
agent=analyst,
tools=[calculator],
context=[gather_task, clean_task],
async_execution=True,
)
```
**Crew** — wires agents and tasks together. Two modes:
```python
# Sequential: DAG topological sort, layered execution, parallel siblings
crew = Crew(agents=[a1, a2, a3], tasks=[t1, t2, t3], process=Process.sequential)
# Hierarchical: a manager agent delegates via LLM
crew = Crew(agents=[a1, a2, a3], tasks=[t1, t2, t3], process=Process.hierarchical)
```
**Flow** — event-driven pipelines. Decorate functions to wire up a DAG. SQLite-backed state persistence with checkpoint/resume.
```python
from mini_crew import Flow
flow = Flow(name="data-pipeline")
@flow.start
def gather():
return fetch_raw_data()
@flow.listen(gather)
def clean(raw):
return preprocess(raw)
@flow.router(clean)
def decide(cleaned):
if cleaned["quality"] > 0.8:
return "publish"
return "review"
@flow.listen("publish")
def publish(data):
return upload(data)
results = flow.run()
```
Flow decorators: `@flow.start` (entry), `@flow.listen(source)` (react to upstream), `@flow.router(source)` (branch to named downstream), `@flow.checkpoint` (human approval gate — resume with `resume_from_checkpoint()`).
## Two engines, pick one
mini-crew gives you two execution engines. They solve different problems.
| Engine | Good for | How it works |
|--------|----------|--------------|
| **CrewAI mode** | Linear pipelines, fixed agent-to-task assignment | Sequential or hierarchical, DAG topo-sort, parallel sibling execution |
| **Flow mode** | Conditional branching, human-in-the-loop, event chains | Event-driven, state graph, SQLite persistence, checkpoint/resume |
```python
from mini_crew import Crew, Engine
# Auto-detect based on task structure
crew = Crew(agents=[...], tasks=[...], engine=Engine.AUTO)
# Or force one
crew = Crew(agents=[...], tasks=[...], engine=Engine.LANGGRAPH)
```
Flow mode uses a LangGraph-style state graph under the hood — conditions, loops, checkpoint nodes. Crew mode is simpler: define agents, assign tasks, pick sequential or hierarchical.
Both share the same Agent/Task/Tool primitives, so switching is cheap.
## Tools & MCP
Define local tools with a decorator:
```python
@tool
def web_search(query: str) -> str:
"""Search the web for information."""
# ... your implementation
```
Tools auto-register in the global `ToolRegistry` and get converted to OpenAI function-calling schemas. Built-in tools: `calculator`, `web_search`, `read_file`, `write_file`, `list_files`, `rag_search`.
### MCP — remote tools like local ones
Expose tools over MCP and consume them from anywhere:
```python
# Server: expose tools
from mini_crew.mcp import MCPServer
server = MCPServer()
server.register_tool(calculator)
server.register_tool(web_search)
server.serve(port=9000)
# Client: discover and use
from mini_crew.mcp import MCPClient
client = MCPClient("http://localhost:9000")
tools = client.discover_tools()
# -> [{"name": "calculator", "schema": {...}}, ...]
agent = Agent(name="Researcher", tools=[local_tool, *tools])
```
MCP tools work identically to local ones — agents don't care where the tool lives.
## Examples
```bash
python examples/travel_planner/main.py # 3-agent sequential crew
python examples/code_review/main.py # 3-agent sequential crew
python examples/market_research/main.py # Flow + LangGraph routing
```
### Travel planner (quick look)
```python
planner = Agent(name="Planner", role="Travel Planner", goal="Design itinerary", tools=[calculator])
reviewer = Agent(name="Reviewer", role="Quality Checker", goal="Verify feasibility")
writer = Agent(name="Writer", role="Report Writer", goal="Produce final document")
plan_task = Task(description="Plan a 3-day trip to Tokyo", agent=planner)
review_task = Task(description="Review the plan for issues", agent=reviewer, context=[plan_task])
write_task = Task(description="Write final itinerary", agent=writer, context=[review_task])
crew = Crew(agents=[planner, reviewer, writer], tasks=[plan_task, review_task, write_task])
crew.kickoff()
```
## Project layout
```
mini-crew/
├── src/mini_crew/
│ ├── agent.py # Agent execution loop + tool calling
│ ├── task.py # Task DAG + topological sort + async parallel
│ ├── crew.py # Crew orchestration (sequential, hierarchical)
│ ├── flow.py # Flow engine: EventBus, SQLite state, checkpoint
│ ├── langgraph_engine.py # LangGraph state graph engine
│ ├── llm.py # LLM adapter (OpenAI SDK → DeepSeek)
│ ├── tool.py # ToolRegistry + @tool decorator
│ ├── mcp/ # MCP server & client
│ ├── types.py # Pydantic models
│ ├── memory.py # Short-term / long-term memory
│ └── utils.py
├── tools/ # Built-in tools
├── examples/ # Demo projects
├── tests/ # 84 tests
└── docs/ # Architecture docs
```
## API at a glance
| Type / Function | What it is |
|----------------|------------|
| `Agent` | LLM worker: role, goal, backstory, tools, `execute()` |
| `Task` | Work unit: description, expected output, agent binding, `async_execution` |
| `Crew` | Orchestrator: `Process.sequential` or `Process.hierarchical`, `kickoff()` |
| `Flow` | Event pipeline: `@start`, `@listen`, `@router`, `@checkpoint` |
| `@tool` | Decorator, auto-registers in ToolRegistry |
| `dag_sort(tasks)` | Topological sort, returns layers for parallel execution |
| `execute_parallel(tasks)` | asyncio parallel execution of same-layer tasks |
| `ToolRegistry` | Global singleton: register, execute, get_schema |
| `MCPClient(url)` | Discover and call remote tools |
| `MCPServer()` | Expose tools over MCP |
| `Process` | Enum: `sequential`, `hierarchical` |
| `Engine` | Enum: `CREWAI`, `LANGGRAPH`, `AUTO` |
| `AgentResult` | content, tool_calls, rounds, success, error |
| `TaskResult` | output, agent_result, success, error |
| `CrewResult` | results, summary, success, total_rounds, total_tool_calls |
## Running tests
```bash
pytest tests/ -v
# 84 passed in ~2s
```
## Tech stack
Python 3.10+ · Pydantic v2 · OpenAI SDK · LangGraph · MCP Protocol · asyncio · loguru · SQLite · pytest
---
*Built by 郭应学 · 江西农业大学 · 2025–2026*
