# 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

mcp-agent-harness/
├── src/
│   └── mcp_agent_harness/
│       ├── __init__.py      # Public exports: AgentHarness, guards, telemetry
│       ├── server.py        # AgentHarness, ToolList, tool decorator, SSE app mounting
│       ├── guards.py        # DataFrame blockade, TokenStarvationError, context & log capping
│       ├── telemetry.py     # Correlated trace_id generation and context logging
│       └── __main__.py      # CLI entrypoint for health checks and standalone daemon
├── tests/
│   ├── test_guards.py       # Unit tests for DataFrame blockade, context truncation, log capping
│   ├── test_server.py       # Sync/async tool execution, leak detection, registry checks
│   └── test_integration_sse.py # FastAPI SSE endpoint mounting and streaming tests
├── scripts/
│   ├── audit_image.sh       # Fail-closed Trivy CVE scanner & prohibited file leak audit
│   └── check_blacklist.sh   # Clean-room sensitive pattern scanner
├── Dockerfile               # Multi-stage hardened Debian container definition
├── docker-compose.yml       # Service definitions for daemon and test runner
└── Makefile                 # Standardized automation targets

from mcp_agent_harness import AgentHarness

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

from mcp_agent_harness import AgentHarness

harness = AgentHarness(name="system-tools")

# Synchronous tool
@harness.tool()
def get_system_load() -> dict:
    return {"load_1m": 0.42, "load_5m": 0.35}

# Asynchronous tool
@harness.tool()
async def fetch_cluster_metrics(region: str) -> dict:
    # Non-blocking async execution
    return {"region": region, "active_nodes": 12, "healthy": True}

import pandas as pd
from mcp_agent_harness import AgentHarness

harness = AgentHarness()

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

# Invoking this tool returns a safe error envelope instead of leaking rows:
# {
#   "status": "error",
#   "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()),
    }

from mcp_agent_harness import truncate_context, cap_log_lines

raw_logs = "INFO: Step completed\n" * 500
# Safe for LLM prompts: capped at 200 lines
clean_logs = cap_log_lines(raw_logs, max_lines=200)

raw_prompt = "User query: analyze output <DATA>large blob...</DATA>"
# Strips bulk data XML tags and caps character length
bounded_prompt = truncate_context(raw_prompt, max_chars=1000)

{
  "status": "error",
  "message": "Division by zero",
  "trace_id": "mcp-e4f912a7"
}

pip install "git+https://github.com/emmanuelol/mcp-agent-harness.git@v0.1.0"

from fastapi import FastAPI
from mcp_agent_harness import AgentHarness

harness = AgentHarness(
    name="production-harness",
    instructions="Production operational agent harness.",
)

@harness.tool()
async def health_summary() -> dict:
    return {"status": "operational", "uptime_pct": 99.98}

app = FastAPI(title="Agent Platform")

@app.get("/health")
def health() -> dict:
    return {"status": "ok"}

# Mount MCP Server-Sent Events endpoint
app.mount("/mcp", harness.sse_app())

uvicorn main:app --host 0.0.0.0 --port 8000