Mantis

AI agent execution engine.

Plug in a model, get a working agent. Tools, search, sandbox, workflows — plug them in, they activate automatically.

from mantis import PipelineExecutor, tool
from mantis.providers import ModelClient

@tool(name="greet", description="Greet someone by name",
      parameters={"name": {"type": "string", "description": "Name to greet"}})
async def greet(name: str) -> dict:
    return {"message": f"Hello {name}"}

executor = PipelineExecutor(
    model=ModelClient(base_url="https://api.openai.com/v1", model="gpt-4o-mini", api_key="sk-..."),
    tools=[greet],
)
result = await executor.run("Say hello to Alice")

Installation

pip install mantis              # core: Executor + @tool + LLM client
pip install mantis[search]      # + graph-tool-call retrieval
pip install mantis[sandbox]     # + Docker sandbox
pip install mantis[state]       # + PostgreSQL checkpointing
pip install mantis[all]         # everything

Only required dependency is httpx.

Architecture

╔══════════════════════════════════════════════════════════════╗
║                     PipelineExecutor                         ║
╠══════════════════════════════════════════════════════════════╣
║                                                              ║
║  ┌─ PREPARE (once) ──────────────────────────────────────┐  ║
║  │  session restore · tool collection · builtin tools     │  ║
║  └────────────────────────────────────────────────────────┘  ║
║       ↓                                                      ║
║  ┌─ RESOLVE (each iteration) ────────────────────────────┐  ║
║  │  tool search (auto) → middleware → LLM call            │  ║
║  └────────────────────────────────────────────────────────┘  ║
║       ↓ if tool_calls                                        ║
║  ┌─ ACT (each iteration) ───────────────────────────────┐  ║
║  │  name correction (auto) → approval → tool execution    │  ║
║  └────────────────────────────────────────────────────────┘  ║
║       ↓                                                      ║
║  ┌─ OBSERVE (each iteration) ───────────────────────────┐  ║
║  │  result feedback → checkpoint (auto) → continue/done   │  ║
║  └────────────────────────────────────────────────────────┘  ║
║       ↓ back to RESOLVE (or done)                            ║
║  cleanup: on_end middleware (always runs)                     ║
║                                                              ║
╠══════════════════════════════════════════════════════════════╣
║  PROVIDERS (plug in to activate)                             ║
║  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐        ║
║  │  Model   │ │  Search  │ │ Sandbox  │ │  State   │        ║
║  │(required)│ │(optional)│ │(optional)│ │(optional)│        ║
║  └──────────┘ └──────────┘ └──────────┘ └──────────┘        ║
╠══════════════════════════════════════════════════════════════╣
║  MIDDLEWARE (cross-cutting concerns)                         ║
║  Approval · Trace · custom                                   ║
╚══════════════════════════════════════════════════════════════╝

Executor Protocol

class Executor(Protocol):
    async def run(self, input: str, *, session_id: str | None = None) -> str: ...
    async def run_stream(self, input: str, *, session_id: str | None = None) -> AsyncIterator[dict]: ...

PipelineExecutor is the default implementation. Implement the Executor protocol to create your own.

Phases

The PipelineExecutor runs 4 phases. Each phase is a Protocol — swap any with your own.

Phase	Runs	What it does
PREPARE	Once per run	Session restore, tool collection, builtin registration
RESOLVE	Each iteration	Tool search (auto), LLM call
ACT	Each iteration	Name correction (auto), approval, tool execution
OBSERVE	Each iteration	Result feedback, checkpoint (auto)

from mantis import PipelineExecutor, PhaseConfig, ExecutionContext
from mantis.executor.phases.protocol import PhaseResult
from mantis.llm.protocol import ModelResponse

class RAGResolve:
    """Inject RAG context before calling LLM."""
    name = "resolve"
    async def run(self, ctx: ExecutionContext) -> PhaseResult:
        rag = await my_rag_search(ctx.run_ctx.last_user_message)
        ctx.conversation.add_system(f"Context: {rag}")
        response = await ctx.model.generate(ctx.conversation.to_messages(), ctx.tools)
        return PhaseResult(data={"response": response})

executor = PipelineExecutor(
    model=ModelClient(...),
    phases=PhaseConfig(resolve=RAGResolve()),
)

Auto-activation

Plug in a provider, it activates automatically:

Provider	What activates
search	Tool filtering in RESOLVE, name correction in ACT, search_tools tool
sandbox	execute_code tool, create_tool tool (with auto-test)
workflows	create_workflow, run_workflow, list_workflows tools
state	Session restore in PREPARE, checkpoint in OBSERVE

Tools

from mantis import tool

@tool(name="lookup_order", description="Look up order by ID",
      parameters={"order_id": {"type": "string", "description": "Order ID"}})
async def lookup_order(order_id: str) -> dict:
    return {"order_id": order_id, "status": "shipped"}

Pass tools as a list:

executor = PipelineExecutor(model=..., tools=[lookup_order, send_slack, query_db])

Streaming

async for event in executor.run_stream("Look up my order"):
    match event["type"]:
        case "thinking":    print(f"thinking... (iter {event['data']['iteration']})")
        case "tool_call":   print(f"calling: {event['data']['name']}")
        case "tool_result": print(f"result: {event['data']}")
        case "done":        print(f"done: {event['data']}")

LLMProvider Protocol

Plug in any LLM:

# Built-in: OpenAI-compatible
executor = PipelineExecutor(model=ModelClient(base_url="...", model="gpt-4o"), ...)

# Custom
class MyLLM:
    async def generate(self, messages, tools=None, temperature=0.7):
        return ModelResponse(text="...", tool_calls=[])

executor = PipelineExecutor(model=MyLLM(), ...)

Middleware

Cross-cutting concerns that must run automatically:

from mantis.middleware import ApprovalMiddleware, TraceMiddleware

executor = PipelineExecutor(
    model=...,
    tools=[...],
    middlewares=[
        TraceMiddleware(),
        ApprovalMiddleware(patterns=["DELETE *", "send_slack"]),
    ],
)

Full Example

from mantis import PipelineExecutor, tool
from mantis.providers import ModelClient, GraphSearch, DockerSandbox, PostgresState
from mantis.workflow import WorkflowStore
from mantis.middleware import ApprovalMiddleware, TraceMiddleware

@tool(name="lookup_order", description="Look up order status by ID",
      parameters={"order_id": {"type": "string", "description": "Order ID"}})
async def lookup_order(order_id: str) -> dict:
    return {"order_id": order_id, "status": "shipped"}

executor = PipelineExecutor(
    model=ModelClient(base_url="https://api.openai.com/v1", model="gpt-4o", api_key="sk-..."),
    tools=[lookup_order],
    search=GraphSearch(),
    sandbox=DockerSandbox(),
    workflows=WorkflowStore(),
    state=PostgresState(dsn="postgresql://..."),
    middlewares=[
        TraceMiddleware(),
        ApprovalMiddleware(patterns=["DELETE *"]),
    ],
)

async for event in executor.run_stream("Build a sales analysis workflow and run it"):
    print(event)

Package Structure

mantis/
├── executor/
│   ├── protocol.py              # Executor Protocol, ExecutionContext, ExecutionEvent
│   ├── pipeline.py              # PipelineExecutor (default implementation)
│   └── phases/                  # PREPARE, RESOLVE, ACT, OBSERVE
│
├── tools/
│   ├── decorator.py             # @tool, ToolSpec
│   ├── registry.py              # ToolRegistry (session scope, source tracking)
│   └── builtins.py              # Auto-registered tools
│
├── providers/                   # Re-exports: ModelClient, GraphSearch, DockerSandbox, ...
├── llm/                         # LLMProvider Protocol + ModelClient
├── middleware/                   # Approval, Trace
├── workflow/                    # WorkflowDef, Store, Runner
├── generate/                    # ToolGenerator
├── sandbox/                     # DockerSandbox
├── search/                      # GraphToolManager
├── context/                     # ConversationContext
├── trace/                       # TraceCollector
├── adapters/                    # SSE adapter, canvas adapter
└── exceptions.py                # MantisError hierarchy

License

MIT