minimal-harness

Documentation: /docs

A lightweight Python agent SDK for building LLM-powered agents with tool-calling support.

Latest version: 0.8.0

Heads up — TUI moved out (0.7.0): The Textual-based TUI client that previously shipped as minimal_harness.client.built_in now lives in its own package: mh-tui. Install it separately with pip install mh-tui. The mh-local CLI command is preserved.

Umbrella: This SDK is one of several packages wired together in the mh-incubator workspace. For the full picture (services, gateway, TUI, frontend) see the umbrella README.

What This Project Is For

Minimal-harness is a lean SDK for building agents that can call tools. It provides:

• OpenAI/Anthropic-compatible API - Works with OpenAI, Anthropic, or any OpenAI-compatible API provider
• Multi-modal image input - Pass image URLs or base64 data to LLM providers supporting vision
• Symmetric Registry + Factory architecture - Register tool/agent metadata with bindings (LocalToolBinding, RemoteToolBinding, ExternalScriptToolBinding); executable instances created lazily by ToolFactory
• Middleware hooks - Observe and intercept the agent lifecycle (agent start/end, LLM calls, tool execution, tool policy enforcement, compaction start/end)
• AsyncIterator events - Real-time async iteration for chunks, tool start/end, execution events, compaction progress
• Conversation memory sessions - Persistent sessions with identity (user_id, scenario_id), auto-persisted to disk
• Auto-compaction - CompactionAgent (agent_type="compacting") folds older messages into a streaming summary whenever the LLM's prompt_tokens exceeds a configured threshold, enabling arbitrarily long conversations
• Remote tools - Pluggable RemoteToolExecutor Protocol; default SSE-over-HTTP executor lives in mh-service-kit
• ESC stop support - Gracefully stop LLM streaming and tool execution

Reference applications

minimal-harness is the SDK. There are four sibling packages that build on it:

Architecture

The SDK is a two-layer framework: a single-shot Agent layer wrapped by a multi-round Controller layer. Controllers add goal-driven / timed orchestration on top of the plain agent loop, without the Agent knowing they exist.

┌────────────────────────────────────────────────────────┐
│ Controller layer   (agent/controller.py)               │
│ default · goal · timer                                  │
│ ControllerStart / ControllerContinue / ControllerEnd   │
├────────────────────────────────────────────────────────┤
│ Agent layer        (agent/*.py)                         │
│ Protocols, types, in-memory primitives                 │
│ Agent loop · Registries · Memory                       │
│ LLM providers · Event types                            │
└────────────────────────────────────────────────────────┘
       │              │          │
       │              │          │
   mh-tui     mh-service-kit    mh-gateway
   (TUI)      (FastAPI service)  (multi-tenant gateway)

Everything above this layer ―sessions, persistence, executors, logging, the TUI, the gateway― lives in the sibling packages.

All event types are defined in src/minimal_harness/types.py. No separate client event layer exists.

Event flow:

async for event in agent.run(
    user_input=[{"type": "text", "text": "..."}],
    memory=memory,
    tools=tools,
):
    if isinstance(event, LLMChunk):
        # handle chunk
    elif isinstance(event, ToolEnd):
        # handle tool result

Controller layer

The Controller protocol (agent/controller.py) wraps an Agent for multi-round orchestration. It has its own event trio:

The SDK ships the framework contract only:

• Controller protocol — execute(agent, user_input, …) signature, identical to Agent.run() plus an agent argument.
• DefaultController (agent/controller.py) — transparent passthrough that always wraps agent.run(). Every runtime run goes through a Controller; there is no "no controller" code path. Unknown controller types in the registry fall back to it.
• ControllerRegistry on AgentRuntime — register(name, factory) lets any app plug in its own controllers; list_types() / catalog() expose them (the gateway's /management/controllers endpoint consumes the catalog).

Application-level controllers (goal / timer loops with a judge LLM) are not in the SDK — they are product policy. mh-gateway implements them in mh_gateway.services.controllers (GoalController, TimerController, shared _LoopingController skeleton) and registers them at startup. They double as the reference sample of an external app plugging a custom controller into the layer.

Selection is per-request: AgentRuntime.run(controller_type=..., controller_config={...}). controller_config (e.g. max_goal_rounds, duration) is passed through to the Controller's execute().

runtime = AgentRuntime(agent_registry=..., session_store=..., ...)
runtime.register_controller(
    "goal", lambda llm_provider: GoalController(llm_provider, max_goal_rounds=5)
)
runtime.register_controller(
    "timer", lambda llm_provider: TimerController(llm_provider, default_duration="30m")
)

task, stop_event, queue = await runtime.run(
    user_input=[{"type": "text", "text": "write 3 poems"}],
    agent_metadata_id="poet",
    memory_id="mem-1",
    controller_type="goal",
    controller_config={"max_goal_rounds": 3},
)

The queue yields AgentEvent | ControllerEvent | None; the final None sentinel marks completion.

Judge safety default: if the judge LLM errors or returns something unparsable, the controller stops (DONE) — it never burns tokens looping on garbage. stop_event is forwarded to the judge call so a user stop interrupts it.

How to Build an App

Project Structure

A typical app looks like this:

my-app/
├── cli.py          # Entry point
└── tools.py        # Your custom tools

1a. Layer 1 — Direct Control

import argparse
import asyncio
from openai import AsyncOpenAI

from minimal_harness.agent.simple import SimpleAgent
from minimal_harness.llm.openai import OpenAILLMProvider
from minimal_harness.memory import ConversationMemory
from minimal_harness.tool.built_in.bash import get_tools as get_bash_tools
from minimal_harness.types import (
    AgentStart,
    AgentEnd,
    LLMChunk,
    ToolStart,
    ToolEnd,
)

def main():
    parser = argparse.ArgumentParser(description="My AI agent")
    parser.add_argument("--base-url", required=True)
    parser.add_argument("--api-key", required=True)
    parser.add_argument("--model", default="deepseek-v4-flash")
    args = parser.parse_args()

    client = AsyncOpenAI(base_url=args.base_url, api_key=args.api_key)
    llm_provider = OpenAILLMProvider(client=client, model=args.model)
    agent = SimpleAgent(llm_provider=llm_provider, max_iterations=50)
    memory = ConversationMemory()
    tools = list(get_bash_tools().values())

    async def run():
        stop_event = asyncio.Event()
        context = {"user_id": "abc123"}  # passed to middleware hooks
        async for event in agent.run(
            user_input=[{"type": "text", "text": "What files are in the current directory?"}],
            stop_event=stop_event,
            memory=memory,
            tools=tools,
            context=context,
        ):
            if isinstance(event, AgentStart):
                print("Agent starting...")
            elif isinstance(event, LLMChunk):
                delta = event.chunk
                if delta and delta.content:
                    print(delta.content, end="", flush=True)
            elif isinstance(event, ToolStart):
                print(f"\n[Calling tool: {event.tool_call['function']['name']}]")
            elif isinstance(event, ToolEnd):
                print(f"\n[Tool result: {str(event.result)[:100]}...]")
            elif isinstance(event, AgentEnd):
                print(f"\n[Done in {event.time_taken:.2f}s]")
                break

    asyncio.run(run())

if __name__ == "__main__":
    main()

1b. Layer 2 — Managed Orchestration

from minimal_harness.agent.runtime import AgentRuntime
from minimal_harness.agent.registry import AgentRegistry
from minimal_harness.tool.registry import ToolRegistry, collect_builtin_tools
from minimal_harness.types import AgentMetadata
from minimal_harness.session import SimpleSession


class InMemorySessionStore:
    """Minimal in-memory session store — replace with your own backend."""

    def __init__(self) -> None:
        self._cache: dict[str, SimpleSession] = {}

    async def create_session(
        self,
        session_id: str | None = None,
        agent_name: str = "",
        user_id: str = "",
        scenario_id: str | None = None,
        transient: bool = False,
        display_name_locale: str | None = None,
    ) -> SimpleSession:
        from uuid import uuid4

        sid = session_id or uuid4().hex
        sess = SimpleSession(
            session_id=sid,
            agent_name=agent_name,
            user_id=user_id,
            scenario_id=scenario_id,
            display_name_locale=display_name_locale,
        )
        self._cache[sid] = sess
        return sess

    async def get_session(self, session_id: str) -> SimpleSession | None:
        return self._cache.get(session_id)

    async def save_memory(self, memory, session_id, extra=None) -> None:
        pass  # in-memory only

    async def delete_session(self, session_id: str) -> bool:
        return self._cache.pop(session_id, None) is not None

    async def list_sessions(self) -> list[dict]:
        return []

    async def list_user_sessions(self, user_id, scenario_id=None) -> list[dict]:
        return []

    async def get_session_messages(self, session_id):
        sess = await self.get_session(session_id)
        return [dict(m) for m in sess.get_all_messages()] if sess else []

    def get_messages_as_items(self, session):
        return [dict(m) for m in session.get_all_messages()]


tool_registry = ToolRegistry()
await collect_builtin_tools(tool_registry)

agent_registry = AgentRegistry()
await agent_registry.register(AgentMetadata(
    name="assistant", display_name="Assistant",
    description="General assistant",
    system_prompt="You are helpful.", agent_type="simple",
    tool_names=["bash", "local_file_operation"],
))

store = InMemorySessionStore()

runtime = AgentRuntime(
    agent_registry=agent_registry,
    session_store=store,
    tool_registry=tool_registry,
    llm_provider_resolver=lambda _: create_llm_provider(...),
)

session = await store.create_session()
task, stop, queue = await runtime.run(
    user_input=[{"type": "text", "text": user_message}],
    agent_metadata_id="assistant",
    memory_id=session.session_id,
)

Note: If you need the handoff and discover_agents runtime tools, they now ship in the mh-tui package as mh_tui.runtime_tools.register_runtime_tools(). They are application glue (multi-agent coordination) rather than core SDK functionality, so they live alongside the TUI that uses them.

2. Add Custom Tools

Tools are defined as async generator functions and registered via ToolMetadata + Binding:

from minimal_harness.tool.registry import ToolRegistry
from minimal_harness.types import ToolMetadata, LocalToolBinding

registry = ToolRegistry()

async def get_weather(location: str) -> AsyncIterator[dict]:
    yield {"success": True, "result": f"The weather in {location} is sunny."}

await registry.register(ToolMetadata(
    name="get_weather",
    display_name="Get Weather",
    description="Get weather for a location",
    parameters={
        "type": "object",
        "properties": {"location": {"type": "string"}},
        "required": ["location"],
    },
    binding=LocalToolBinding(fn=get_weather),
))

Or use the @register_tool decorator (recommended pattern — omit registry and call register_decorated_tools() during async setup):

from minimal_harness.tool.registration import register_tool, register_decorated_tools

@register_tool(
    name="get_weather",
    description="Get weather for a location",
    parameters={
        "type": "object",
        "properties": {"location": {"type": "string"}},
        "required": ["location"],
    },
    # registry=...  # optional — see below
)
async def get_weather(location: str) -> AsyncIterator[dict]:
    yield {"success": True, "result": f"The weather in {location} is sunny."}

# Later, during async setup:
await register_decorated_tools(registry)

For remote tools, use RemoteToolBinding:

from minimal_harness.types import RemoteToolBinding

await registry.register(ToolMetadata(
    name="weather",
    description="Get weather",
    parameters={...},
    binding=RemoteToolBinding(url="https://my-service.com/weather"),
))

For external script tools, use ExternalScriptToolBinding:

from minimal_harness.types import ExternalScriptToolBinding

await registry.register(ToolMetadata(
    name="my_tool",
    description="...",
    parameters={...},
    binding=ExternalScriptToolBinding(script_path="/path/to/tool.py"),
))

Localized tool output: Tools can detect the user's language at runtime via get_current_locale():

from minimal_harness.agent.runtime import get_current_locale

async def my_tool() -> AsyncIterator[dict]:
    locale = get_current_locale()
    yield {"message": "你好" if locale == "zh" else "Hello"}

3. Run

python cli.py --base-url https://api.openai.com/v1 --api-key sk-... --model gpt-4o

Or set environment variables:

export MH_BASE_URL=https://api.openai.com/v1
export MH_API_KEY=sk-...
export MH_MODEL=gpt-4o
python cli.py

Middleware Hooks

Subclass Middleware to observe or intercept the agent lifecycle:

from minimal_harness.agent.middleware import Middleware
from minimal_harness.types import LLMEnd, ToolCall

class PolicyEnforcer(Middleware):
    async def should_allow_tool(
        self, tool_call: ToolCall, **kwargs
    ) -> bool | str:
        if tool_call["function"]["name"] == "bash":
            return "bash is not permitted in this context"
        return True

    async def on_llm_end(self, event: LLMEnd) -> None:
        if event.usage:
            print(f"Tokens: {event.usage['total_tokens']}")

Pass middleware to SimpleAgent:

agent = SimpleAgent(
    llm_provider=llm_provider,
    middleware=[PolicyEnforcer()],
    max_iterations=50,
)

Multi-modal Image Input

Pass image URLs or base64-encoded image data as input content parts:

user_input = [
    {"type": "text", "text": "What's in this image?"},
    {
        "type": "image",
        "image_url": {"url": "https://example.com/photo.jpg"},
    },
]

For local images, encode as base64:

import base64

with open("photo.jpg", "rb") as f:
    data = base64.b64encode(f.read()).decode()

user_input = [
    {"type": "text", "text": "Describe this image"},
    {
        "type": "image",
        "data": data,
        "media_type": "image/jpeg",
    },
]

Built-in Tools

The SDK ships no tools of its own. The bash and local_file_operation tools live in mh-tui as mh_tui.built_in (they're application-level concerns that the TUI happens to ship). To use them outside the TUI, copy the module — it's about 400 lines and depends only on minimal_harness.tool.base and minimal_harness.types.

from mh_tui.built_in import collect_builtin_tools, get_tools

# Register them into a ToolRegistry in one call
await collect_builtin_tools(tool_registry)  # → set[str] of names

# Or use the Tool instances directly
for name, tool in get_tools().items():
    print(name, tool.display_name)

Event Types

All events are defined in minimal_harness.types and consumed as a single AgentEvent union:

LLMChunkDelta contains content, reasoning, and tool_calls fields for provider-agnostic partial deltas.

Message IDs & Event Ordering

Every message that enters a session receives a canonical id (msg-{seq}, per-session counter) stamped by Memory.add_message at insert time. The id is persisted in MemoryData (next_message_seq), survives compaction and tool-message discards, and is the same value read-side session adapters return after a reload — streaming consumers can therefore commit feedback/references against an id that will still match after a refresh.

Two ordering contracts are important for consumers:

1. MessageEvent is emitted before LLMEnd for the same turn. The agent persists the turn's messages (reasoning/assistant) into memory first, then broadcasts LLMEnd carrying the assistant message's message_id. Consumers must not assume LLMEnd arrives before that turn's MessageEvent.
2. Memory.add_message mutates the passed dict — it stamps id in place (idempotent: an existing id is left untouched). Callers must not rely on the dict being unmodified, and should copy it (e.g. dict(message)) before broadcasting a MessageEvent if they need the pre-stamp view.

AgentEnd.message_id is the id of the last assistant message of the run (or None if the run produced none); LLMEnd.message_id is the id of that turn's assistant message; CompactionEnd.message_id is the id of the summary message (or None when compaction failed). Tool messages do not have a ToolEnd.message_id — tool results are persisted only after all tools finish (declaration order, for LLM context stability), so the id is delivered by the MessageEvent(role="tool") that follows.

Batch Evaluation

The minimal_harness.eval module has been removed. Use the mh-gateway's eval API or POST /api/v1/eval/batch to run agent evaluation campaigns.

Environment Variables

The SDK no longer reads environment variables. The MH_* env vars are read by their respective consumers:

Stop Mechanism

Pass an asyncio.Event to agent.run(..., stop_event=event) and event.set() it from any concurrent task (e.g. an HTTP handler, a key press handler) to gracefully stop LLM streaming and tool execution. The TUI (mh-tui) wires this to the Esc key.