How to Build a Custom Agent Harness

How to Build a Custom Agent Harness

Building a custom agent harness is a practical way to orchestrate AI agents, manage their tools, and control their interactions. Unlike off-the-shelf frameworks, a custom harness gives you fine-grained control over agent behavior, context handling, and error recovery. This article walks through a complete, hands-on implementation using Python, focusing on clarity and extensibility.

Requirements

Before starting, ensure your environment meets these requirements:

• Python 3.10 or later installed on your system
• pip package manager (comes with Python)
• Basic familiarity with command-line tools and Python syntax
• An API key for at least one LLM provider (OpenAI is used in examples)

The following Python packages are needed:

• `openai` – for LLM API calls
• `httpx` – for HTTP requests to tools
• `pydantic` – for data validation and configuration

Step-by-step Installation

1. Set up a Python virtual environment

Isolate dependencies to avoid conflicts with other projects.

python -m venv agent-harness
source agent-harness/bin/activate  # On Windows: agent-harness\Scripts\activate

2. Install required packages

Install the core libraries for API communication and data modeling.

pip install openai httpx pydantic

3. Create the project structure

Organize your code into modules for maintainability.

mkdir -p agent_harness/tools
touch agent_harness/__init__.py
touch agent_harness/core.py
touch agent_harness/tools/__init__.py
touch agent_harness/tools/search.py
touch agent_harness/tools/calculator.py
touch main.py

4. Configure environment variables

Store your API keys securely. Create a `.env` file (not committed to version control) or export them directly.

export OPENAI_API_KEY="your-api-key-here"

Building the Core Agent Harness

The Agent Class

The heart of the harness is an `Agent` class that manages the LLM, tools, and conversation history.

# agent_harness/core.py
import os
from typing import List, Dict, Any, Callable
from openai import OpenAI

class Tool:
    """Represents a tool the agent can use."""
    def __init__(self, name: str, description: str, function: Callable):
        self.name = name
        self.description = description
        self.function = function

class Agent:
    def __init__(self, model: str = "gpt-4", tools: List[Tool] = None):
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.model = model
        self.tools = tools or []
        self.history = []
        
    def add_tool(self, tool: Tool):
        """Register a new tool."""
        self.tools.append(tool)
        
    def build_system_prompt(self) -> str:
        """Create system prompt listing available tools."""
        tool_descriptions = "\n".join(
            [f"- {t.name}: {t.description}" for t in self.tools]
        )
        return f"""You are a helpful assistant with access to the following tools:
{tool_descriptions}

When you need to use a tool, respond with:
TOOL: tool_name
ARGS: arg1=value1, arg2=value2

Otherwise, respond directly."""

Adding Tool Execution

The harness must parse tool calls and execute them safely.

# agent_harness/core.py (continued)
    def run(self, user_input: str) -> str:
        """Process user input and return agent response."""
        self.history.append({"role": "user", "content": user_input})
        
        # Build messages for LLM
        messages = [{"role": "system", "content": self.build_system_prompt()}]
        messages.extend(self.history)
        
        # Get LLM response
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=0.7
        )
        
        content = response.choices[0].message.content
        
        # Check for tool invocation
        if content.startswith("TOOL:"):
            tool_name = content.split("\n")[0].replace("TOOL:", "").strip()
            args_line = content.split("\n")[1].replace("ARGS:", "").strip()
            args = self._parse_args(args_line)
            
            # Execute tool
            for tool in self.tools:
                if tool.name == tool_name:
                    result = tool.function(**args)
                    self.history.append({"role": "assistant", "content": content})
                    self.history.append({"role": "tool", "content": str(result)})
                    return self.run(f"Tool result: {result}")
                    
        self.history.append({"role": "assistant", "content": content})
        return content
        
    def _parse_args(self, args_str: str) -> Dict[str, Any]:
        """Parse key=value arguments from string."""
        args = {}
        for pair in args_str.split(","):
            key, value = pair.split("=")
            args[key.strip()] = value.strip()
        return args

Building Example Tools

Web Search Tool

A simple tool that performs a mock web search. In production, replace with an actual API.

# agent_harness/tools/search.py
import httpx

def web_search(query: str) -> str:
    """Perform a simulated web search."""
    # In production, use a real search API like SerpAPI or Bing
    return f"Simulated search results for '{query}': Found 3 relevant documents."

Calculator Tool

A safe calculator using Python's `eval` with restricted globals.

# agent_harness/tools/calculator.py
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression safely."""
    allowed_names = {"abs", "round", "min", "max", "sum", "pow"}
    try:
        # Only allow basic math operations
        result = eval(expression, {"__builtins__": {}}, {name: __builtins__.__dict__[name] for name in allowed_names})
        return f"Result: {result}"
    except Exception as e:
        return f"Error: {str(e)}"

Usage Examples

Running the Harness

Create a main script to demonstrate the agent in action.

# main.py
from agent_harness.core import Agent, Tool
from agent_harness.tools.search import web_search
from agent_harness.tools.calculator import calculator

def main():
    # Initialize agent
    agent = Agent(model="gpt-4")
    
    # Register tools
    agent.add_tool(Tool(
        name="web_search",
        description="Search the internet for information",
        function=web_search
    ))
    agent.add_tool(Tool(
        name="calculator",
        description="Evaluate mathematical expressions",
        function=calculator
    ))
    
    # Interactive loop
    print("Agent Harness Ready. Type 'exit' to quit.")
    while True:
        user_input = input("\nYou: ")
        if user_input.lower() == "exit":
            break
            
        response = agent.run(user_input)
        print(f"Agent: {response}")

if __name__ == "__main__":
    main()

Run the harness:

python main.py

Example interaction:

Agent Harness Ready. Type 'exit' to quit.

You: What is 2 + 2?
Agent: 4

You: Search for latest AI news
Agent: Simulated search results for 'latest AI news': Found 3 relevant documents.

Advanced Configuration

Add support for conversation memory and tool timeout.

# agent_harness/core.py (add to Agent class)
    def clear_history(self):
        """Reset conversation context."""
        self.history = []
        
    def run_with_timeout(self, user_input: str, timeout: int = 30) -> str:
        """Execute with a timeout for safety."""
        import signal
        
        def handler(signum, frame):
            raise TimeoutError("Agent execution timed out")
            
        signal.signal(signal.SIGALRM, handler)
        signal.alarm(timeout)
        
        try:
            return self.run(user_input)
        finally:
            signal.alarm(0)

Best Practices

Error Handling

Wrap tool execution in try-except blocks to prevent agent crashes.

# In Agent.run()
try:
    result = tool.function(**args)
except Exception as e:
    result = f"Tool execution failed: {str(e)}"

Rate Limiting

Implement exponential backoff for API calls.

import time
from openai import RateLimitError

def run_with_retry(self, user_input: str, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            return self.run(user_input)
        except RateLimitError:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)

Logging

Add structured logging for debugging and monitoring.

import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# In Agent.run()
logger.info(f"User input: {user_input}")
logger.info(f"Tool invoked: {tool_name}")

Conclusion

Building a custom agent harness gives you complete control over how your AI agents operate. This implementation provides a solid foundation with:

• A modular architecture that separates core logic from tools
• Safe tool execution with error handling
• Clean conversation history management
• Easy extensibility for new capabilities

The harness can be extended with features like parallel tool execution, streaming responses, or persistent state. Start with this base, then add tools specific to your domain—whether it's database queries, API integrations, or custom data processing. The key is to maintain the separation between the agent's reasoning and tool execution, which makes debugging and scaling straightforward.

As you build more complex agents, remember to prioritize safety through input validation, timeout mechanisms, and resource monitoring. A well-built harness will serve as the reliable backbone for your AI applications, whether you're powering a chatbot, a research assistant, or an automated workflow system.