HarnessAgent

Production-grade, multi-tenant agent harness for building, running, observing, and self-improving AI agents. HarnessAgent wraps any agent framework (LangGraph, CrewAI, AutoGen) or can run native SQL/Code agents with a full production lifecycle: paged context management, semantic memory, tool safety, distributed tracing, and automated self-improvement.

What it gives you

Architecture

HarnessAgent is structured as a layered service. External requests enter through the FastAPI layer and pass through the orchestration layer to individual agents, which interact with memory, tools, and observability systems.

Quick Start

Install

pip install harnessagent[vector,observe,mcp]
# Or with all extras:
pip install harnessagent[all]bash

Run a single native agent

from harness.core.context import AgentContext
from harness.agents.sql_agent import SQLAgent
from harness.memory.manager import MemoryManager
from harness.llm.router import LLMRouter
from pathlib import Path

# Build components
memory = await MemoryManager.create(config)
llm    = LLMRouter()
llm.register(anthropic_provider, priority=0)

agent = SQLAgent(
    llm_router=llm,
    memory_manager=memory,
    tool_registry=tool_registry,
    safety_pipeline=None,
    step_tracer=None,
    mlflow_tracer=None,
    failure_tracker=failure_tracker,
    audit_logger=audit_logger,
    event_bus=event_bus,
    cost_tracker=cost_tracker,
    checkpoint_manager=checkpoint_manager,
    trace_recorder=trace_recorder,   # ← new: durable span tree
)

ctx = AgentContext.create(
    tenant_id="acme",
    agent_type="sql",
    task="List all tables and their row counts",
    memory=memory,
    workspace_path=Path("/workspaces/acme/run1"),
)

result = await agent.run(ctx)
print(result.output, result.cost_usd, result.steps)python

Wrap an existing LangGraph agent

import harness
from langgraph.graph import StateGraph

graph = StateGraph(...)  # your existing graph
adapter = harness.wrap(graph)
adapter.attach_harness(
    safety_pipeline=my_pipeline,
    cost_tracker=cost_tracker,
    audit_logger=audit_logger,
)

async for event in adapter.run_with_harness(ctx, {"input": "analyze sales data"}):
    print(event.event_type, event.payload)python

Start the API server

uvicorn harness.api.main:create_app --factory --host 0.0.0.0 --port 8000
# Or via Makefile:
make apibash

Configuration

All settings are loaded from environment variables or a .env file via harness.core.config.Settings (Pydantic BaseSettings). Access the singleton via get_config().

Core — Context & Events

Defined in harness/core/context.py. These are the fundamental data objects that flow through every part of the system.

AgentContext

Mutable per-run state shared across all components. Created once per run and passed everywhere.

StepEvent

Emitted at each significant moment in an agent run. Published to EventBus (Redis Pub/Sub) for SSE streaming.

Core — Protocols

Defined in harness/core/protocols.py. Structural Protocol ABCs that all pluggable components must satisfy. Use isinstance(x, LLMProvider) for runtime checking.

Core — Error Hierarchy

All harness errors extend HarnessError and carry a canonical FailureClass enum value. The failure class drives HTTP status codes, Prometheus labels, and Hermes sampling.

BaseAgent — Run Lifecycle

Defined in harness/agents/base.py. All concrete agents inherit from BaseAgent. The run(ctx) method orchestrates the full production lifecycle.

Constructor Parameters

Run loop steps

1. Open RUN trace span, emit started event, begin MLflow run
2. Resume from checkpoint if one exists (_maybe_resume_checkpoint)
3. Loop while ctx.is_budget_ok():
   1. Fit history to context window (compress if > 40 messages)
   2. GraphRAG/vector retrieve — build retrieval context string
   3. Build LLM messages + system prompt
   4. Open LLM trace span → call LLM → register token usage → close span
   5. Open GUARDRAIL span → run safety pipeline → close span
   6. If no tool calls → extract final answer → break
   7. HITL approval check (sequential)
   8. Execute all tool calls in parallel, each in a TOOL span
   9. Checkpoint every 10 steps
4. Close RUN span (OK or ERROR), emit completed/failed event
5. Return AgentResult

Methods to override in subclasses

CodeAgent

harness/agents/code_agent.py — Specialized for Python code writing, debugging, and iterative improvement.

agent_type: "code"

Workflow: Understand → Write clean Python with type hints → Lint with ruff → Run code → Iterate on failures.

Default tools: run_python, lint_code, read_file, write_file, apply_patch, list_workspace

SQLAgent

harness/agents/sql_agent.py — Specialized for SQL queries, schema exploration, and data analysis.

agent_type: "sql"

Schema pre-population: Before the main loop, _populate_schema(ctx) introspects the database and stores table/column nodes in the knowledge graph. This enables GraphRAG to surface schema context automatically in subsequent queries.

Default tools: list_tables, describe_table, sample_rows, execute_sql

Framework Adapters

harness/adapters/ — Wraps external agent frameworks with the harness production lifecycle.

import harness

# Auto-detect from object type
adapter = harness.wrap(my_langgraph_graph)   # → LangGraphAdapter
adapter = harness.wrap(my_crew)               # → CrewAIAdapter
adapter = harness.wrap(my_autogen_agent)      # → AutoGenAdapter

# Inject production components
adapter.attach_harness(safety_pipeline, cost_tracker, audit_logger)
adapter.attach_mcp(mcp_client, tool_names=["search", "execute"])python

MemoryManager

harness/memory/manager.py — Unified interface to all memory tiers. Injected into every agent via AgentContext.memory.

Memory tiers

Factory

memory = await MemoryManager.create(config, llm_provider=llm)
# Builds: ShortTermMemory + EmbeddingProvider + VectorStore
#         + GraphMemory + ContextWindowManager + ContextEnginepython

Key methods

ContextEngine

harness/memory/context_engine.py — Paged, skill-isolated context management with automatic offload and semantic retrieval.

Offload pipeline

1. Check hot window size (O(1) Redis LLEN)
2. If count < _HOT_MAX_MSGS (200), skip
3. Compute hot token usage (scan all raw items)
4. If < 80% of max_hot_tokens, skip
5. Walk from oldest to newest until ~2 000 tokens collected
6. Compress (LLM preferred, extractive fallback)
7. Embed summary → store in vector DB with metadata
8. Store ContextPage in Redis HASH + ZSET (scored by importance)
9. LTRIM hot list to remove offloaded messages

Importance scoring

Cold pages with higher importance are kept longer. Score starts at 0.5 and bumps for:

• +0.10 tool messages
• +0.15 error/failed content
• +0.08 result/found/success content

Action scoring formula

composite = 0.5 × goal_progress
          + 0.3 × tool_relevance
          + 0.2 × confidence

goal_progress  = min(1, keyword_overlap(response, goal) × 2)
tool_relevance = 0.9 (success + result > 20 chars)
               | 0.6 (success, no result)
               | 0.5 (no tool)
               | 0.1 (error)
confidence     = 0.6 − 0.08 per hedging phrase
               + 0.08 per affirming phraseformula

Key constants

GraphRAG Engine

harness/memory/graph_rag.py — Weighted multi-hop graph retrieval with vector bridging. Used by SQLAgent (schema) and any knowledge-intensive workload.

Edge weights

Retrieval strategy (priority order)

1. Extract entities from query (regex patterns for table/column names)
2. Find matching graph nodes (find_nodes(entities))
3. If no anchors found → vector-to-graph bridging: embed query, find vector hits, use as graph seeds
4. BFS from anchor nodes, score paths by cumulative edge weight
5. Render top-N paths as compact context: [SCHEMA], [JOINS], [PAST QUERIES], [PAST ERRORS]
6. Supplement with raw vector hits if graph coverage is thin

Semantic LLM Cache

harness/llm/cache.py — Cosine-similarity cache for LLM responses, keyed by message embedding.

Trace System — Overview

HarnessAgent produces a hierarchical span tree for every agent run. Spans are stored in Redis (48 h TTL) and appended to logs/runs/{run_id}/trace.jsonl.

Span hierarchy

The span stack is tracked automatically per run_id inside TraceRecorder._stacks. Callers never need to pass parent_span_id — it is always the top of the stack.

TraceSpan Schema

harness/observability/trace_schema.py

TraceRecorder

harness/observability/trace_recorder.py

Persistence sinks

Usage

recorder = TraceRecorder.create(redis_url=cfg.redis_url, log_dir="logs")

# Context manager (recommended)
async with recorder.span(run_id, SpanKind.TOOL, "tool:execute_sql", ctx,
                          input_preview=str(args)) as span_id:
    result = await registry.execute(ctx, call)

# Manual control
span_id = await recorder.start_span(run_id, SpanKind.LLM, "llm:call", ctx)
recorder.set_llm_usage(span_id, input_tokens=200, output_tokens=100, cost_usd=0.003)
await recorder.end_span(run_id, span_id)

# Query
trace = await recorder.get_trace(run_id)   # → TraceView | None
span  = await recorder.get_span(span_id)   # → TraceSpan | Nonepython

Metrics & Audit

Prometheus metrics (harness/observability/metrics.py)

AuditLogger (harness/observability/audit.py)

Append-only log for compliance. Writes to {workspace_base}/audit/{tenant_id}/{date}.jsonl and Redis stream harness:audit.

Actions tracked: tool_call, memory_write/read, llm_call, agent_start/complete, hitl_request/resolve, safety_block, patch_apply

PII safety: payloads are SHA-256 hashed before storage — content is never stored, only fingerprints.

Event Bus (SSE)

harness/observability/event_bus.py — Redis Pub/Sub channels for real-time streaming via Server-Sent Events.

Channel: harness:events:{run_id}

# BaseAgent emits
await event_bus.publish(StepEvent.llm_called(ctx, response))

# Client subscribes (used by SSE route)
async for event in event_bus.subscribe(run_id):
    print(event.event_type, event.payload)

# From the browser:
const source = new EventSource(`/runs/${runId}/stream`)
source.onmessage = (e) => console.log(JSON.parse(e.data))python / js

ToolRegistry

harness/tools/registry.py — Central tool management with validation, safety, timeout, and audit.

Execution pipeline

1. Lookup — find tool by name; raise TOOL_NOT_FOUND if missing
2. Schema validation — validate args against input_schema; raise TOOL_SCHEMA_ERROR if invalid
3. Safety check — run through safety_pipeline; raise SAFETY_STEP if blocked
4. Timeout — wrap execution in asyncio.timeout(tool.timeout_seconds)
5. Execute — call tool.execute(ctx, args)
6. Audit log — record in AuditLogger
7. Metrics — increment tool_calls_total

Built-in tools

MCP Client

harness/tools/mcp_client.py — Connect to any MCP-compatible server (stdio or SSE) and expose its tools in the ToolRegistry.

from harness.tools.mcp_client import MCPToolAdapter, load_mcp_servers_from_config

# From YAML config
adapters = load_mcp_servers_from_config("mcp_servers.yaml")
for adapter in adapters:
    tools = await adapter.connect()
    for tool in tools:
        registry.register(tool)

# Manual
adapter = MCPToolAdapter(MCPServerConfig(
    name="search",
    transport="stdio",
    command=["npx", "-y", "@modelcontextprotocol/server-brave-search"],
    env={"BRAVE_API_KEY": "${BRAVE_API_KEY}"},
))
tools = await adapter.connect()python

YAML config format

servers:
  search:
    transport: stdio
    command: ["npx", "-y", "@modelcontextprotocol/server-brave-search"]
    env: {BRAVE_API_KEY: "${BRAVE_API_KEY}"}
  filesystem:
    transport: stdio
    command: ["npx", "-y", "@modelcontextprotocol/server-filesystem", "/data"]yaml

Skills

harness/tools/skills.py — Named, versioned, composable agent capabilities. Skills bundle a system prompt fragment, required tools, and usage examples.

from harness.tools.skills import Skill, SkillRegistry

registry = SkillRegistry()
registry.register(Skill(
    name="sql_analysis",
    version="1.0.0",
    description="Analyze SQL databases",
    system_prompt="You are an expert SQL analyst...",
    required_tools=["list_tables", "describe_table", "execute_sql"],
    tags=["sql", "analytics"],
))

# Compose multiple skills
combined = registry.compose("sql_analysis", "data_visualization")

# Filter by tag
analytics_skills = registry.list_for_tags(["analytics"])python

Execution Sandbox

harness/filesystem/sandbox.py — Isolated code execution. Docker is the primary sandbox; RestrictedPython is the fallback.

DockerSandbox security controls

AgentRunner

harness/orchestrator/runner.py — Creates, executes, and tracks agent runs. The main component called by API routes.

RunRecord states

Planner & Scheduler

Multi-agent DAG decomposition and parallel execution.

Planner (harness/orchestrator/planner.py)

Takes a complex task, calls an LLM to decompose it into a JSON array of SubTask objects with dependency declarations. Validates the DAG (cycle detection via Kahn's algorithm).

planner = Planner(llm_provider=llm)
plan = await planner.plan(
    task="Build a sales dashboard: fetch data, analyze it, generate report",
    available_agents=["sql", "code", "research"],
)
# Returns TaskPlan with SubTasks:
# {id: "fetch", agent_type: "sql",  depends_on: []}
# {id: "analyze", agent_type: "code", depends_on: ["fetch"]}
# {id: "report",  agent_type: "code", depends_on: ["analyze"]}python

Scheduler (harness/orchestrator/scheduler.py)

Executes the plan by repeatedly finding ready tasks (deps satisfied) and running them in parallel with asyncio.gather + semaphore back-pressure.

• Parallelism: max_concurrent=10 semaphore
• Retry: max_retries=1 per subtask, exponential backoff
• Handoff enrichment: predecessor outputs injected into dependent task context
• Failure propagation: tasks with failed deps are skipped and marked failed

Human-in-the-Loop (HITL)

harness/orchestrator/hitl.py — Pause agent execution and wait for human approval before executing sensitive tool calls.

# In agent context metadata
ctx.metadata["hitl_manager"] = hitl_manager
ctx.metadata["policy"] = policy  # policy.requires_hitl(tool_name) → bool

# During tool execution, BaseAgent calls:
request = await hitl_manager.request_approval(ctx, tool_name, tool_args)
decision = await hitl_manager.await_decision(request.request_id, timeout=3600.0)
# "approved" → continue; "rejected"/"expired" → raise HITLRejectedpython

Pending requests are stored in Redis sorted set harness:hitl:pending and exposed via GET /hitl/pending.

Inter-Agent Messaging

harness/messaging/bus.py — Redis Streams-based messaging between agents in multi-agent systems.

LLM Router

harness/llm/router.py — Health-aware, context-window-aware routing across multiple LLM providers with per-provider circuit breaking.

Routing algorithm

1. Sort registered providers by priority (lower = higher priority)
2. Skip providers where context_window < required_context
3. Skip providers with open circuit breakers
4. Try each provider in priority order
5. On retryable error (LLM_RATE_LIMIT, LLM_TIMEOUT, LLM_ERROR): try next provider
6. Return response from first successful provider

Circuit breaker

Registering providers

from harness.llm.router import LLMRouter
from harness.llm.anthropic import AnthropicProvider
from harness.llm.openai_provider import OpenAIProvider

router = LLMRouter()
router.register(AnthropicProvider(api_key=cfg.anthropic_api_key),
                priority=0, context_window=200_000)
router.register(OpenAIProvider(api_key=cfg.openai_api_key, model="gpt-4o-mini"),
                priority=10, context_window=128_000)
# Falls back to OpenAI if Anthropic is unavailablepython

Hermes — Self-Improvement Loop

Hermes is the self-healing system that automatically detects failure patterns, generates prompt patches, evaluates them, and optionally applies them — with regression detection and auto-rollback.

Scoring formula

score = success_rate
      − 0.01 × avg_steps_delta
      − 0.001 × avg_tokens_deltaformula

Regression detection (OnlineLearningMonitor)

After every run, metrics are recorded per prompt version. If the post-patch error count exceeds baseline × (1 + regression_threshold), the patch is automatically rolled back to the previous version.

Evaluation Framework

harness/eval/runner.py and harness/eval/diagnostics.py

from harness.eval.runner import EvalRunner
from harness.eval.datasets import EvalDataset, EvalCase

dataset = EvalDataset(
    name="sql_smoke",
    agent_type="sql",
    cases=[
        EvalCase(case_id="c1", task="List all tables", expected_output="users, orders, products"),
        EvalCase(case_id="c2", task="Count users", expected_output="42"),
    ],
)

runner = EvalRunner(agent_runner=agent_runner)
report = await runner.run(dataset, concurrency=3)
print(report.success_rate, report.avg_cost_usd)
print(report.to_markdown())python

EvalDiagnostics

Each EvalReport includes rich per-case diagnostics:

• failure_stage — where the agent failed: llm | tool | safety | timeout | budget | quality
• Recommendations — auto-generated hints based on failure patterns
• By-agent aggregates — token usage, cost, tool errors, guardrail hits per agent type

REST API Reference

Runs

Traces

Evaluations

Improvement

Authentication

Pass an API key in the X-API-Key header. In dev mode, anonymous requests default to tenant "default". JWT bearer tokens are also supported.

Run response schema

{
  "run_id":       "3f2a8c1e4d...",
  "tenant_id":    "acme",
  "agent_type":   "sql",
  "task":         "List all tables",
  "status":       "completed",
  "created_at":   "2024-01-01T00:00:00Z",
  "completed_at": "2024-01-01T00:00:12Z",
  "result": {
    "output":          "Found 7 tables: users, orders...",
    "steps":           4,
    "tokens":          1250,
    "cost_usd":        0.00031,
    "success":         true,
    "elapsed_seconds": 11.8
  }
}json

Trace response schema

{
  "trace_id":             "ddda858ebe8f...",
  "run_id":               "3f2a8c1e4d...",
  "agent_type":           "sql",
  "status":               "ok",
  "duration_ms":          11840,
  "total_input_tokens":   980,
  "total_output_tokens":  270,
  "total_cost_usd":       0.00031,
  "span_count":           6,
  "spans": [
    {
      "span_id":        "01f04e9413851d7f",
      "parent_span_id": null,
      "kind":           "run",
      "name":           "run:sql",
      "status":         "ok",
      "duration_ms":    11840,
      "input_preview":  "List all tables",
      "output_preview": "Found 7 tables..."
    },
    ...
  ]
}json

Redis Schema

Dependencies

Core (always installed)

Optional extras