# MCP Agent Harness

[![CI Pipeline](https://github.com/emmanuelol/mcp-agent-harness/actions/workflows/ci.yml/badge.svg)](https://github.com/emmanuelol/mcp-agent-harness/actions/workflows/ci.yml)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)
[![Python: 3.11+](https://img.shields.io/badge/python-3.11+-blue.svg)](https://www.python.org/downloads/)
[![Release: v0.2.0](https://img.shields.io/badge/release-v0.2.0-green.svg)](https://github.com/emmanuelol/mcp-agent-harness/releases)

Generic, SRE-hardened Model Context Protocol (MCP) harness for autonomous agents. Acts as an operational firewall preventing LLM context exhaustion, enforcing data boundary blockades, throttling runaway agent retry loops, and guaranteeing zero unhandled panics across SSE streams.

---

## 5-Line FastAPI Quickstart

```python
from fastapi import FastAPI
from mcp_agent_harness import AgentHarness

app = FastAPI()
harness = AgentHarness("ops-agent")
@harness.tool(max_chars=1000, rate_limit=(10, 60.0), catch_unhandled=True)
def get_status() -> str: return "System healthy"
app.mount("/mcp", harness.sse_app())
```

---

## Architecture: The MCP SRE Firewall

```mermaid
flowchart TD
    Client["Autonomous LLM / MCP Client (OpenClaw)"]
    SSE["FastAPI Mount (/mcp/sse)"]
    
    subgraph Harness ["AgentHarness SRE Firewall"]
        Trace["1. Telemetry Tracer (init_trace)"]
        Rate["2. Sliding Window Rate Limiter"]
        Func["3. Tool Execution (Sync/Async)"]
        DFGuard["4. DataFrame Blockade (Zero-Import MRO)"]
        LogGuard["5. Log Line Capping (cap_log_lines)"]
        CtxGuard["6. Context Truncation & XML Strip"]
        ErrWrap["7. Deterministic Error Envelope"]
    end
    
    Backend["Host Backend Services (DB, Redis, APIs)"]

    Client -->|JSON-RPC via SSE| SSE
    SSE --> Trace
    Trace --> Rate
    Rate --> Func
    Func <--> Backend
    Func --> DFGuard
    DFGuard --> LogGuard
    LogGuard --> CtxGuard
    CtxGuard -->|Clean Bounded Payload| Client
    Func -.->|Unhandled Panic / Rate Limit| ErrWrap
    DFGuard -.->|TokenStarvationError| ErrWrap
    ErrWrap -->|JSON Error Envelope| Client
```

---

## Benchmark Metrics & SRE Hardening

| Metric | Unguarded FastMCP | AgentHarness v0.2.0 | Operational Impact |
| :--- | :--- | :--- | :--- |
| **Token Consumption (1k rows)** | ~85,000 tokens | **~420 tokens** | **99.5% prompt token reduction** |
| **Prompt Injection Defense** | Raw XML `<DATA>` leak | **Sanitized scalar preview** | Immune to XML unclosed tag context hijack |
| **Hallucination Loop Defense** | Uncapped infinite retries | **Sliding Window Rate Limit** | Hard stop on agent runaway retry storms |
| **Stream Stability** | SSE drops on unhandled panic | **100% Stream Uptime** | Deterministic JSON error envelope return |
| **Guard Latency Overhead** | None | **< 0.15 ms** | Sub-millisecond boundary verification |
| **Dependencies** | Requires `pandas` at install | **Zero Core C-deps** | MRO duck-typing for pandas, polars, pyarrow |

---

## Prerequisites

- **Docker** `>= 24.0`
- **Docker Compose** `>= 2.20`
- **GNU Make** `>= 4.3`
- *(Optional for local development)* **Python** `>= 3.11`

---

## Quickstart (Containerized)

No local Python environment or dependency installation required. All lifecycle steps run inside isolated containers:

```bash
# 1. Build the hardened container image
make build

# 2. Run the integration test suite
make test

# 3. Spin up the harness and verify live endpoint health
make demo

# 4. Run the security, leak, and Trivy CVE audit gate
make audit
```

---

## API & Configuration Reference

### `AgentHarness`

Main server class wrapping FastMCP with SRE boundary guards.

```python
from mcp_agent_harness import AgentHarness

harness = AgentHarness(
    name="analytics-mcp",
    instructions="Operational analytics tools for agents.",
    register_builtins=True,
)
```

| Parameter | Type | Default | Description |
| :--- | :--- | :--- | :--- |
| `name` | `str` | `"mcp-agent-harness"` | Identifier for the MCP server instance. |
| `instructions` | `str` | `"Generic SRE-hardened MCP Harness"` | System instructions exposed to MCP clients. |
| `register_builtins` | `bool` | `True` | Registers default diagnostic tools (`health`, `ping`, `diagnostics`). |

### Declarative Tool Decorator (`@harness.tool`)

Configurable declarative SRE policies applied at the decorator boundary:

```python
@harness.tool(
    max_chars=2000,
    max_lines=200,
    strip_xml=True,
    catch_unhandled=True,
    rate_limit=(10, 60.0),
)
def query_system_logs(service: str) -> str:
    ...
```

| Parameter | Type | Default | Description |
| :--- | :--- | :--- | :--- |
| `max_chars` | `Optional[int]` | `None` | Automatically truncates string outputs exceeding character limit. |
| `max_lines` | `Optional[int]` | `None` | Caps string output lines to prevent context terminal flooding. |
| `strip_xml` | `bool` | `True` | Strips raw `<TAG>...</TAG>` blocks when `max_chars` is enforced. |
| `catch_unhandled` | `bool` | `False` | Catches unexpected exceptions and returns structured JSON error envelopes. |
| `rate_limit` | `Optional[tuple[int, float]]` | `None` | `(max_calls, period_seconds)` sliding window limiter per tool. |

---

## Guardrails in Action

### 1. Token Starvation Guard (DataFrame Blockade)

The harness blocks raw tabular data from entering LLM context. Returning a raw DataFrame immediately triggers `TokenStarvationError`, captured into a structured error envelope:

```python
import pandas as pd
from mcp_agent_harness import AgentHarness

harness = AgentHarness()

# ❌ BAD: Leaks raw DataFrame
@harness.tool()
def query_orders() -> pd.DataFrame:
    return pd.DataFrame({"order_id": [1, 2], "amount": [99.5, 14.2]})

# Invoking returns a safe error envelope instead of leaking rows:
# {
#   "status": "error",
#   "error_code": "TokenStarvationError",
#   "message": "Raw DataFrames are strictly prohibited in MCP tools. Return scalar metrics or token-compressed strings.",
#   "trace_id": "mcp-3a1b4c9e"
# }

# ✅ GOOD: Return compact scalar metrics
@harness.tool()
def query_order_metrics() -> dict:
    df = pd.DataFrame({"order_id": [1, 2], "amount": [99.5, 14.2]})
    return {
        "order_count": len(df),
        "total_revenue": float(df["amount"].sum()),
        "average_order": float(df["amount"].mean()),
    }
```

### 2. Hallucination Throttling (Rate Limiting)

Throttles runaway LLM retries when an agent enters an infinite retry loop:

```python
@harness.tool(rate_limit=(3, 60.0), catch_unhandled=True)
def trigger_external_mutation(action: str) -> dict:
    return {"status": "dispatched", "action": action}

# On the 4th invocation within 60s, execution is stopped before hitting backend:
# {
#   "status": "error",
#   "error_code": "ToolRateLimitError",
#   "message": "Rate limit exceeded: 3 calls per 60.0s limit reached.",
#   "trace_id": "mcp-9b2f11c4"
# }
```

---

## Telemetry & Error Envelopes

Every tool invocation automatically initializes an OpenTelemetry-compatible `trace_id` (`mcp-xxxxxxxx`).

When `catch_unhandled=True` is configured, internal tool exceptions are safely converted to structured responses without crashing the SSE connection:

```json
{
  "status": "error",
  "error_code": "ConnectionResetError",
  "message": "Remote database closed connection",
  "trace_id": "mcp-e4f912a7"
}
```