⠒

Agent Harness is Fundamentally Simple

Agent harnesses come with goals, subagents, planners, and much more. Feels quite complex, doesn’t it?

Fundamentally, however, they don’t have to be. Pi, my daily driver, builds on this idea: minimalism, also in agent harnesses.

I like stripping things down to their core - microai project, is one example. When I recently needed to build an agent (more on that in another post), my first instinct was to train a model. A harness around an existing one proved a better starting point.

What Is a Harness?

A harness is the software around an LLM that lets it carry out a task: instructions, tools, conversation, and the machinery connecting them. Here’s a nice article with more detail.

Take a coding agent. You ask it to fix a bug. It reads the source, runs a test, sees the failure, edits the code, and tries again. Each result becomes context for the next decision.

There are surprisingly few moving pieces:

  1. An LLM: decides what to do, given instructions and context.
  2. Tools: execute the doing part; named operations with defined arguments and results.
  3. A loop: enables repetition; tool results feed back into the model to support the next decision.
  4. Events: communicate progress and outcomes; updates let the rest of the application follow along.

The Architecture

Those pieces make the core. To turn it into a coding agent, we add an orchestrator that supplies concrete tools, prepares context, and manages the task. This is the business logic.

Around it sit the interfaces: APIs, TUIs, web UIs. Different ways to use the same orchestrator, sharing the same logic.

The Core

The core understands messages and tool calls. A Provider streams responses; a ToolRuntime describes and executes tools. Both are interfaces. The loop only knows how to invoke them.

Below is a simplified core; imports are omitted for readability.

loop.py - the model/tool exchange
class AgentLoop:
    def __init__(self, provider: Provider) -> None:
        self.provider = provider

    async def run(self, context: AgentLoopContext, tools: ToolRuntime) -> AsyncIterator[AgentEvent]:
        for _ in range(context.max_steps):
            assistant = None
            try:
                async for event in self.provider.stream(context.provider_messages, tools.definitions()):
                    yield event
                    if isinstance(event, CompletionEnd):
                        assistant = event.message
            except Exception as error:
                yield AgentError(str(error))
                return

            if assistant is None:
                yield AgentError("Provider ended without a complete message.")
                return

            context.provider_messages.append(assistant)
            if not assistant.tool_calls:
                yield AgentCompleted(assistant)
                return

            for call in assistant.tool_calls:
                yield ToolCallStart(call)
                try:
                    result = await tools.execute(call)
                except Exception as error:
                    result = ToolResult(str(error), is_error=True)
                yield ToolCallEnd(call, result)
                if result.requires_input:
                    yield InputRequired()
                    return
                context.provider_messages.append(tool_message(call, result))

        yield MaxStepsReached()


def tool_message(call: ToolCall, result: ToolResult) -> ProviderMessage:
    return ProviderMessage(
        role="tool",
        content=f"{"error" if result.is_error else "success"}: {result.content}",
        tool_call_id=call.id,
        image_data_urls=(result.image_data_url,) if result.image_data_url else (),
    )

The return trip matters. Results go back into the conversation, paired with their call IDs. Errors become feedback, too. Results can include images or request user input.

A response without tool calls finishes the loop. A step budget bounds model turns; tool implementations still need their own timeouts.

schemas.py - messages and interfaces
@dataclass
class ToolDefinition:
    name: str
    description: str
    parameters: dict[str, object]


@dataclass
class ToolCall:
    id: str
    name: str
    arguments: dict[str, object] = field(default_factory=dict)


@dataclass
class ToolResult:
    content: str
    is_error: bool = False
    requires_input: bool = False
    image_data_url: str | None = None


class ToolRuntime(Protocol):
    def definitions(self) -> Sequence[ToolDefinition]: ...

    async def execute(self, call: ToolCall) -> ToolResult: ...


@dataclass
class ProviderMessage:
    role: Literal["system", "user", "assistant", "tool"]
    content: str = ""
    tool_calls: list[ToolCall] = field(default_factory=list)
    tool_call_id: str | None = None
    image_data_urls: tuple[str, ...] = ()

@dataclass
class AgentLoopContext:
    provider_messages: list[ProviderMessage] = field(default_factory=list)
    max_steps: int = 32


class Provider(Protocol):
    def stream(self, messages: Sequence[ProviderMessage], tools: Sequence[ToolDefinition]) -> AsyncIterator[ProviderEvent]: ...

The provider emits start/delta/end events for completions and, when available, reasoning. The loop adds tool and lifecycle events.

events.py - following the work
@dataclass
class CompletionStart:
    pass


@dataclass
class CompletionDelta:
    text: str


@dataclass
class CompletionEnd:
    message: ProviderMessage
    duration_ms: int


@dataclass
class ReasoningStart:
    pass


@dataclass
class ReasoningDelta:
    text: str


@dataclass
class ReasoningEnd:
    reasoning: str
    duration_ms: int


ProviderEvent = (
    CompletionStart
    | CompletionDelta
    | CompletionEnd
    | ReasoningStart
    | ReasoningDelta
    | ReasoningEnd
)


@dataclass
class ToolCallStart:
    call: ToolCall


@dataclass
class ToolCallEnd:
    call: ToolCall
    result: ToolResult


@dataclass
class AgentCompleted:
    message: ProviderMessage


@dataclass
class InputRequired:
    pass


@dataclass
class MaxStepsReached:
    pass


@dataclass
class AgentError:
    message: str


AgentEvent = (
    ProviderEvent
    | ToolCallStart
    | ToolCallEnd
    | AgentCompleted
    | InputRequired
    | MaxStepsReached
    | AgentError
)

The Orchestrator

Now we give the loop a job. A coding agent, for example, needs a workspace and concrete tools such as file read or edit. The orchestrator supplies these implementations and prepares the conversation around the task.

It also owns the task’s lifecycle: retaining conversation history, handling user interactions, and checking the result.

The application boundary - an interface, too
@dataclass
class TaskRequest:
    instruction: str
    workspace_root: Path


@dataclass
class TaskEvent:
    kind: Literal["working", "input_required", "completed", "failed"]
    message: str = ""


class Orchestrator(Protocol):
    def run(self, request: TaskRequest) -> AsyncIterator[TaskEvent]: ...

The orchestrator owns the business logic and exposes it through operations and events. An Agent Client Protocol (ACP) adapter can offer these to ACP-compatible clients.

The Interfaces

Put a TUI on it, or expose it over HTTP for a web UI. Both send requests to the same orchestrator and turn its events into feedback. A task update might become an API message or a terminal redraw.

Dependencies point inward, like an onion: interfaces -> orchestrator -> core.

This is the software-design part I enjoy: giving each piece one job and a boundary that makes sense. Change a provider, add a terminal interface, test the loop with a fake tool - the pieces fit.

A small core doesn’t make the surrounding work disappear. It supports everything we build around it.