This is my attempt at a long form post about some of the things I have learned by building my own agent.
Structure of an agent
What is an agent? There are plenty of definitions available, but for now we're just going to say
An agent is a ReAct Loop + Tools
At the heart of every agent harness (we use the terms interchangeably) is the ReAct Loop.
This is how every coding agent works. They take your message, add it to a conversation history and send that to an LLM. Then the react loop program parses the output from the LLM. If there are tool calls, the program executes the tool calls on your machine. Whatever else an agent harness is doing, this is the core execution model of agents in November 2026.
Building on top of ReAct
So a ReAct loop is not all that an agent requires. The main components of an agent are as follows
- LLM Provider
- Configuration
- Persistence
- ReAct Loop
- Tools
- Client
Language Choice
I highly recommend python for developing your agent harness. So you hate python and want to avoid it entirely? That's great! Here's why you should still use python for your react loop + agent core.
- The React loop is completely bottle necked on LLM API response time
- You gain nothing by using a AOT compiled language
- The libraries in Python for every step of this are extremely mature
- You are going to sacrifice a huge amount of development speed by insisting on using another language, even a language I like, like Julia
- You can probably use typescript instead honestly, but I still recommend python just because
I know the pull to use rust is great, but trust me on this you're going to be paying on compilation times to gain effectively nothing in terms of performance.
LLM Provider
Just use openai package. Whatever you do, don't start using litellm as a package. Set it up as a proxy if you absolutely must, but never import litellm in your harness code that is the way of pain. If you just say "fuck anthropic" and embrace chat/completions API from llama.cpp you'll save yourself a word of effort and let yourself focus on the things that actually matter.
Configuration
Honestly this is probably the most important part of building an agent harness. You want to use good dataclasses for the configuration and make as much as possible configurable. I prefer YAML to store my configurations but JSONC is also a viable alternative. I would keep my configurations in ~/.agents/{your_harness_name}/config.yaml. And I would also keep the secrets in ~/.agents/{your_harness_name}/.env and do whatever os permissions you need to keep your API keys secret.
Configuration is going to be a huge part of your harness so it is very good to have a plan in mind before you start writing the rest of your specs on how you want your agent built. You almost can't overthink your configuration, because it can get to be a bit of a hornet's nest if you aren't careful with it.
Persistence
So you want to have the capability to load a previous session and speak to it. A lot of harnesses use jsonl in their configuration directory to save message history and it makes sense. You're just sending JSON back and forth to the LLM so just write it in a file.
This is a mistake. While it is difficult to completely map the branchy JSON of tool calls to SQL, you don't have to. You can store a mixture of JSON and non JSON data in SQL easily and define your entire data model for your app's persistence in SQLAlchemy. This will let you switch between postgres and sqlite with almost trivial changes to your code to enable both.
Images
If you want a real harness that's worth its salt, you have to support images. I am not going to go into the exact structure of openai chat completions images, but you need to save these and what you don't want is to save them in sql. You're going to have a memory tool at some point and it's going to pull that blob in and have 2M tokens sent to the LLM endpoint lol.
So you save your images to file for local version and hydrate the data contained in sql when you send to LLM from local blobs.
Before minio went non-open source, it was the clear winner in terms of distributed file servers. Now you want to set it up to where in your configurations you can give it an address for an S3 compatible RustFS endpoint that you can run yourself.
So yeah images are a bit of a bear, but they're absolutely worth it because without them you're stuck reading text. So bit the bullet and plan to support images in your harness from day one.
ReAct Loop
Here's a react loop in less than 300 lines of python
import asyncio
import json
import os
import sys
from dataclasses import dataclass, field
from pathlib import Path
import httpx
from fastmcp import Client
from mcp.types import ImageContent
from openai import APIConnectionError, AsyncOpenAI, RateLimitError
from openai._exceptions import APITimeoutError
API_KEY = os.getenv("ALIBABA_API_KEY")
BASE_URL = os.getenv("ALIBABA_BASE_URL", "https://token-plan.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1")
MODEL = os.getenv("REACT_MODEL", "qwen3.8-max")
MAX_TOKENS = 8192
CANCELLED_MESSAGE = "Tool call cancelled by user"
MCP_CONFIG = {
"mcpServers": {
"crow-mcp": {
"transport": "stdio",
"command": str(Path.home() / ".agents/crow/src/crow-cli/.venv/bin/crow-cli"),
"args": ["mcp2"],
}
}
}
def configure_provider(raw: bool = False) -> AsyncOpenAI:
async def log_request(request):
print(f"\n{'=' * 20} RAW REQUEST {'=' * 20}\n{request.read().decode()}\n")
hooks = {"request": [log_request]} if raw else None
return AsyncOpenAI(
api_key=API_KEY,
base_url=BASE_URL,
http_client=httpx.AsyncClient(event_hooks=hooks) if hooks else None,
)
async def get_tools(mcp_client: Client) -> list[dict]:
return [
{
"type": "function",
"function": {"name": t.name, "description": t.description or "", "parameters": t.inputSchema},
}
for t in await mcp_client.list_tools()
]
async def send_request(lm, messages, model, tools, max_retries=3):
for attempt in range(max_retries):
try:
return await lm.chat.completions.create(
model=model,
messages=messages,
tools=tools,
stream=True,
max_tokens=MAX_TOKENS,
stream_options={"include_usage": True},
temperature=0.6,
parallel_tool_calls=False,
)
except (APITimeoutError, RateLimitError, APIConnectionError):
if attempt == max_retries - 1:
raise
await asyncio.sleep(2**attempt)
@dataclass
class StreamState:
thinking: list[str] = field(default_factory=list)
content: list[str] = field(default_factory=list)
tool_calls: dict[int, dict] = field(default_factory=dict) # by delta index
usage: dict | None = None
def reset(self) -> None:
self.thinking.clear()
self.content.clear()
self.tool_calls.clear()
self.usage = None
def accumulate_chunk(chunk, state: StreamState) -> list[tuple[str, object]]:
if getattr(chunk, "usage", None) is not None:
state.usage = {
"prompt_tokens": chunk.usage.prompt_tokens,
"completion_tokens": chunk.usage.completion_tokens,
"total_tokens": chunk.usage.total_tokens,
}
if not chunk.choices:
return []
delta = chunk.choices[0].delta
tokens: list[tuple[str, object]] = []
reasoning = getattr(delta, "reasoning_content", None)
if reasoning:
state.thinking.append(reasoning)
tokens.append(("thinking", reasoning))
if delta.content:
state.content.append(delta.content)
tokens.append(("content", delta.content))
for call in delta.tool_calls or ():
slot = state.tool_calls.setdefault(call.index, {"id": "", "function_name": "", "arguments": []})
if call.id:
slot["id"] = call.id
if call.function and call.function.name:
slot["function_name"] = call.function.name
tokens.append(("tool_call", (call.function.name, call.function.arguments or "")))
if call.function and call.function.arguments:
slot["arguments"].append(call.function.arguments)
if not any(kind == "tool_call" for kind, _ in tokens):
tokens.append(("tool_args", call.function.arguments))
return tokens
def build_tool_calls(state: StreamState) -> list[dict]:
calls = []
for _, slot in sorted(state.tool_calls.items()):
arguments = "".join(slot["arguments"])
try:
json.loads(arguments)
except (json.JSONDecodeError, TypeError, ValueError):
arguments += "}" * (arguments.count("{") - arguments.count("}"))
arguments += "]" * (arguments.count("[") - arguments.count("]"))
try:
json.loads(arguments)
except (json.JSONDecodeError, TypeError, ValueError):
arguments = "{}"
calls.append(
dict(id=slot["id"], type="function",
function=dict(name=slot["function_name"], arguments=arguments))
)
return calls
def deserialize_args(data):
if isinstance(data, str):
if data.startswith(("{", "[")):
try:
return deserialize_args(json.loads(data))
except (json.JSONDecodeError, TypeError, ValueError):
pass
return data
if isinstance(data, dict):
return {k: deserialize_args(v) for k, v in data.items()}
if isinstance(data, list):
return [deserialize_args(item) for item in data]
return data
def mcp_to_llm(mcp_content) -> object:
if not mcp_content:
return ""
blocks = []
for item in mcp_content:
if isinstance(item, ImageContent):
blocks.append({"type": "image_url", "image_url": {"url": f"data:{item.mimeType};base64,{item.data}"}})
else:
blocks.append({"type": "text", "text": str(getattr(item, "text", item))})
if len(blocks) == 1 and blocks[0]["type"] == "text":
return blocks[0]["text"]
return blocks
async def execute_tool_calls(mcp_client, tool_call_inputs, tool_results: list[dict]):
try:
for tool_call in tool_call_inputs:
name = tool_call["function"]["name"]
raw_args = tool_call["function"]["arguments"]
args = deserialize_args(raw_args)
if not isinstance(args, dict):
tool_call["function"]["arguments"] = "{}" # in place: history is re-sent
content = (f"Error: malformed arguments for '{name}': {raw_args!r}. "
"Retry with valid JSON matching the tool schema.")
else:
try:
result = await mcp_client.call_tool_mcp(name, args)
content = mcp_to_llm(result.content)
if result.isError:
content = f"Error: {content}"
except Exception as exc:
content = f"Error: {exc}"
tool_results.append(dict(role="tool", tool_call_id=tool_call["id"], content=content))
return tool_results
except asyncio.CancelledError:
responded = {r["tool_call_id"] for r in tool_results}
for tool_call in tool_call_inputs:
if tool_call["id"] not in responded:
tool_results.append(
dict(role="tool", tool_call_id=tool_call["id"], content=CANCELLED_MESSAGE)
)
raise
def add_response_to_messages(messages, state, tool_call_inputs, tool_results):
if not (state.content or state.thinking or tool_call_inputs):
return messages # an empty completion persists nothing
thinking = "".join(state.thinking)
msg = {"role": "assistant", "content": "".join(state.content)}
if thinking:
msg["reasoning_content"] = thinking
if tool_call_inputs:
msg["tool_calls"] = tool_call_inputs
messages.append(msg)
messages.extend(tool_results)
return messages
async def react_loop(messages, mcp_client, lm, model, tools, max_turns=50000):
state = StreamState()
for _ in range(max_turns):
response = await send_request(lm, messages, model, tools)
state.reset()
try:
async for chunk in response:
for kind, token in accumulate_chunk(chunk, state):
yield {"type": kind, "token": token}
except asyncio.CancelledError:
calls = [c for c in build_tool_calls(state) if c["id"] and c["function"]["name"]]
add_response_to_messages(
messages, state, calls,
[dict(role="tool", tool_call_id=c["id"], content=CANCELLED_MESSAGE) for c in calls],
)
raise
tool_call_inputs = build_tool_calls(state)
if not tool_call_inputs:
add_response_to_messages(messages, state, [], [])
yield {"type": "usage", "token": state.usage}
return
tool_results: list[dict] = []
try:
await execute_tool_calls(mcp_client, tool_call_inputs, tool_results)
except asyncio.CancelledError:
add_response_to_messages(messages, state, tool_call_inputs, tool_results)
raise # placeholders already filled for every unanswered call
add_response_to_messages(messages, state, tool_call_inputs, tool_results)
def make_renderer():
last = None
def render(chunk):
nonlocal last
kind, token = chunk["type"], chunk["token"]
if last == "tool_args" and kind != "tool_args":
print(")")
if kind in ("thinking", "content"):
if last != kind:
print(("\n" if last else "") + f"[{kind}] ", end="")
print(token, end="", flush=True)
elif kind == "tool_call":
print(f"\n[tool] {token[0]}({token[1]}", end="", flush=True)
elif kind == "tool_args":
print(token, end="", flush=True)
elif kind == "usage":
print(f"\n[usage] {token}")
last = kind
return render
async def run(cancel_after: float | None, raw: bool) -> list[dict]:
mcp_client = Client(MCP_CONFIG)
lm = configure_provider(raw)
render = make_renderer()
messages = [
dict(role="system", content="You are a helpful assistant named Crow."),
dict(role="user", content="search for machine learning papers with your search tool"),
]
async with mcp_client:
tools = await get_tools(mcp_client)
if cancel_after:
task = asyncio.current_task()
asyncio.get_running_loop().call_later(cancel_after, task.cancel)
print(f"[cancelling in {cancel_after}s]")
try:
async for chunk in react_loop(messages, mcp_client, lm, MODEL, tools):
render(chunk)
except asyncio.CancelledError:
# The point of the exercise: history is still valid to resume from.
# Every assistant tool_calls entry has a matching tool response.
print(f"\n[cancelled — {len(messages)} messages in history]")
for msg in messages[2:]:
print(" ", json.dumps(msg, ensure_ascii=False)[:220])
return messages
if __name__ == "__main__":
argv = sys.argv[1:]
cancel = float(argv[argv.index("--cancel-after") + 1]) if "--cancel-after" in argv else None
asyncio.run(run(cancel, "--raw" in argv))Tools
As you can see from the above example, you should just use MCP for your tools. Why reinvent the wheel? Huge reason to use python for your harness is the stability of FastMCP. Don't reinvent a new way of doing tools in a react loop. Just use MCP. It's the standard format for tool calling with LLMs.
When you are designing future systems with your harness, adding new tools is going to be "create an MCP that does yada yada" and it is going to make your life so much easier than having the agent need to load a whole skill on your custom, bespoke tool calling bullshit that your agent vibe coded together. Don't write your own YAML parser, and use MCP for tool calling.
Client
Lot of people are use to use coding agents through TUIs, so when you talk about client they're like "Client? What the fuck are you talking about client? The agent is a program you run from command line? There's no client."
And to those people I must preach the glory and majesty of Agent-Client Protocol.
If you adopt ACP early and vigorously it will collapse all sorts of uncertainty around how your agent is supposed to behave. Just like you shouldn't reinvent the wheel when you are tool calling with your react loop, you shouldn't try to reinvent the wheel of user-agent interaction that the ACP specifications exist to standardize and codify.
HIGHLY recommend thinking about the full structure of the ACP specs (minus tool permissions personally, but you do you) before you start designing your agent's data structure.
For one you can use your agent inside any number of official ACP clients like Zed or any JetBrains IDE. But more importantly, it gives you as an agent harness developer options for how you want to interact with your agent. Build one agent server and use your agent from the web with assistant-ui or from the command line with a TUI via Martty. You can build ACP clients into any sort of interface you can imagine and plug your agent server in easy as pie.
So adopting ACP gives you a critical separation of concerns that I don't honestly understand how other agent platforms really operate without.
Build an ACP agent server. Then you can put it on the registry and everyone will know your name.
Shameless self-plug
If you want to get started building agents immediately instead of focusing on reinventing the wheel, I highly recommend investigating building custom agents in python with crow-cli. All you have to do to build your own agent in crow-cli is define:
- A
SystemPromptFactorycallable which returns a jinja2 template and a callable to generate the args for the template - An optional Compaction callable to customize compaction
from __future__ import annotations
import argparse
import asyncio
import re
from pathlib import Path
from typing import Any, Callable, Iterable, Optional
from crow_cli.agent.prompt import SystemPromptResponse, default_system_prompt
from crow_cli.agent2.main import agent_run
from crow_cli.config import Config
SystemPromptFactory = Callable[[Config, str, Optional[str]], SystemPromptResponse]
ROLE = """
* Your primary role is to DO the work: explore, implement, test, and commit it.
* You are one step in a sequence. A reviewer that never saw your work reads your
final message and tries to break every claim in it, so a claim you did not
check is a defect you are handing downstream.
* If you are asked a question ("why is X happening"), answer it; do not fix it.
"""
REPORT = """
Your final message is a HANDOFF, not a summary for a human who was watching.
Whoever reads it has no memory of your work and no way to ask a follow-up:
whatever is not in that message did not happen.
DONE - <what you changed, file by file>
VERIFIED - <the exact command you ran> -> <what it returned>
NOT DONE - <what you left, and why>
COMMIT - <hash and subject>, or "uncommitted: <why>"
Every line under VERIFIED carries the command that produced it. "Tests pass" is
a claim; "`uv run pytest -q` -> 3 passed" is evidence. Commit before you report.
Call no tools after you start writing the handoff -- it is the last thing you say.
"""
def patch_section(template: str, name: str, body: str) -> str:
pattern = re.compile(rf"^(<{name}>$)(.*?)(^</{name}>$)", re.M | re.S)
if not pattern.search(template):
raise KeyError(f"system prompt has no <{name}> section to patch")
return pattern.sub(lambda m: f"{m.group(1)}\n{body.strip()}\n{m.group(3)}", template, count=1)
def insert_section(template: str, name: str, body: str, *, after: str) -> str:
pattern = re.compile(rf"^</{after}> CODEBLOCK1 quot;, re.M)
if not pattern.search(template):
raise KeyError(f"system prompt has no </{after}> to insert <{name}> after")
block = f"</{after}>\n\n<{name}>\n{body.strip()}\n</{name}>"
return pattern.sub(lambda _m: block, template, count=1)
def role_prompt(
patches: Optional[dict[str, str]] = None,
extra: Optional[Iterable[tuple[str, str, str]]] = None,
*,
role: str = "crow-role",
) -> SystemPromptFactory:
def factory(config: Config, cwd: str, session_id: Optional[str] = None) -> SystemPromptResponse:
base = default_system_prompt(config, cwd, session_id)
template = base.template
for name, body in (patches or {}).items():
template = patch_section(template, name, body)
for name, body, after in extra or ():
template = insert_section(template, name, body, after=after)
args: dict[str, Any] = {**base.template_args, "crow_role": role}
return SystemPromptResponse(template=template, template_args=args)
return factory
system_prompt = role_prompt(
patches={"ROLE": ROLE},
extra=[("REPORT", REPORT, "IMPORTANT")],
role="worker",
)
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
parser.add_argument("--http", action="store_true", help="serve Streamable HTTP + WebSocket instead of stdio")
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=2780)
parser.add_argument("--model", default=None, help="override the config's model for every session")
parser.add_argument("--config-dir", type=Path, default=None)
parser.add_argument("--config-file", type=Path, default=None, help="yaml overrides, e.g. a scratch db_uri")
parser.add_argument("--debug", action="store_true", help="log every request and stream chunk to disk")
args = parser.parse_args()
asyncio.run(
agent_run(
system_prompt=system_prompt,
http=args.http,
host=args.host,
port=args.port,
model=args.model,
config_dir=args.config_dir,
config_file=args.config_file,
debug=args.debug,
)
)
if __name__ == "__main__":
main()Compaction
I guess I'm tired of writing, but one of the most important aspects of your harness is going to be how you handle compaction. Do you write notes in some file and instruct new agents to read? Do you do like I do and just have agent summarize the work of the session and point RESTfully at crucial files while describing context for next agent? If you use crow-cli as a dependency, you can do anything you want with compaction without ever modifying the source code.
If you fear compaction, you should probably write your own harness. I did and I fear no compaction event.