mcp-harness

Enterprise governance middleware for MCP servers. Cost attribution, auth, observability, quotas, audit, and policy — composable around the official MCP Python SDK.

Why this exists

Most public MCP servers are toy examples. The gap between "works on my laptop" and "I can ship this to 500 engineers under SOC 2, GDPR, and a Finance team that wants per-business-unit cost allocation" is enormous — and almost none of that gap is in the protocol. It's in the boring middleware around it.

mcp-harness is that middleware. It doesn't fork the protocol, doesn't reinvent the SDK, and doesn't try to be an agent framework. It's a set of composable decorators and middleware that wrap the official SDK so the resulting server can be safely run inside a company.

The headline use case: your CFO asks who's spending the AI budget, and you can answer by end of week one.

Install

pip install mcp-harness            # core pipeline (no MCP SDK required)
pip install 'mcp-harness[server]'  # + the official MCP SDK, to actually serve tools
pip install 'mcp-harness[all]'     # + OpenTelemetry, Prometheus, tiktoken, JWT

The core middleware pipeline has no hard dependency on mcp, so you can unit-test all your governance behaviour without a transport. Install the server extra to run a live server.

Hello, governed world

The same MCP server you'd write anyway, with the governance layer declared once at the top:

from mcp_harness import Harness
from mcp_harness.auth import APIKeyAuth
from mcp_harness.governance import CostTracking, Quotas, AuditLog
from mcp_harness.observability import OTELTracing, StructuredLogging
from mcp_harness.policy import AllowList

harness = Harness(
    name="customer-data-mcp",
    auth=APIKeyAuth(keys={"sk-finance": {"id": "svc-reports", "team": "finance"}}),
    middleware=[
        OTELTracing(service_name="customer-data-mcp"),
        StructuredLogging(),
        AllowList.from_yaml("policies/tool-access.yaml"),
        Quotas(per_principal_per_minute=60),
        CostTracking(
            cost_center_resolver=lambda p: p.team,
            sink="jsonl://costs.jsonl",
        ),
        AuditLog(sink="jsonl://audit.jsonl"),
    ],
)

@harness.tool()
async def search_customer(customer_id: str) -> dict:
    """Find a customer record by ID."""
    return {"id": customer_id, "name": "ACME Corp"}

if __name__ == "__main__":
    harness.run()  # stdio by default; transport="streamable-http" for HTTP

Already have a server? Wrap it in one line

Keep every @server.tool() exactly as it is — from_fastmcp adopts an existing FastMCP instance and routes its tools through the pipeline, preserving their schemas:

from mcp.server.fastmcp import FastMCP
from mcp_harness import Harness
from mcp_harness.governance import CostTracking, Quotas

server = FastMCP("my-mcp")

@server.tool()
async def search(q: str) -> dict:
    return {"hits": []}

harness = Harness.from_fastmcp(server, middleware=[CostTracking(), Quotas(per_principal_per_minute=60)])
harness.run()

Or scaffold it from the CLI, without touching your file:

mcp-harness wrap server.py     # writes governed_server.py next to it
mcp-harness init my-mcp        # scaffold a fresh governed server

Then attribute spend:

$ mcp-harness daily-rollup costs.jsonl
cost_center     calls      in_tok     out_tok      cost_usd
----------------------------------------------------------
finance            42        5210         9830        0.1284
platform           18        1940         3110        0.0451
----------------------------------------------------------
TOTAL              60        7150        12940        0.1735

How it works

A tiny onion middleware pipeline, decoupled from the SDK. @harness.tool() registers your function; calls flow through each layer and into your tool. The same path runs whether the call arrives from a live MCP client or from the in-process test client.

client ─▶ FastMCP ─▶ wrapper ─┐
                              ├─▶ auth ─▶ policy ─▶ quotas ─▶ tracing ─▶ cost ─▶ audit ─▶ your tool
test/direct ─▶ dispatch ──────┘

Each layer is independently useful, independently testable, and opt-in. Adopt just CostTracking, or stack the whole thing.

Modules

Optional integrations degrade gracefully: OTELTracing is a no-op (warned once) without the otel extra; Metrics falls back to an in-memory backend without prometheus-client; cost token counting uses a heuristic without tiktoken; RedisQuotaStore and SchemaGuard raise a clear HarnessError pointing at the right extra ([redis] / [schema]) if used without it installed.

Run mcp-harness doctor to see which of those are active in your environment:

$ mcp-harness doctor
mcp-harness 0.2.0
python      3.12.0 (/usr/local/bin/python)

  [x] server   mcp 1.28.1
  [ ] otel     -- OTELTracing span export
  [x] metrics  prometheus-client 0.21.1
  [ ] tokens   -- exact token counting (else a chars/4 heuristic)

Shutting down cleanly

AuditLog writes through a background queue, and sinks may hold files or sockets. Closing the harness drains every layer that needs it, so nothing buffered is lost on exit:

async with Harness(name="my-mcp", middleware=[AuditLog(sink="jsonl://audit.jsonl")]) as harness:
    ...  # serve, or drive it from tests

# or explicitly, e.g. in a `finally:` around harness.run()
await harness.aclose()

Governance never fails a call it didn't mean to reject: a sink that can't write warns and drops the record instead of turning a successful tool call into an error, and HTTPSink retries connection failures, timeouts, 429s, and 5xxs before giving up.

Testing your server

Governance is covered by ordinary unit tests — no transport, no mocks of the SDK:

from mcp_harness.testing import HarnessTestClient, MockPrincipal
from myserver import harness

async def test_finance_can_search(harness_client):  # fixture auto-registered
    client = harness_client(harness, principal=MockPrincipal("svc-a", team="finance"))
    assert await client.call("search_customer", {"customer_id": "c-1"})

Examples

• examples/quickstart_server.py — smallest governed server.
• examples/local_governed_server.py — full stack, zero cloud deps. Run python examples/local_governed_server.py --demo to watch the governance layers reject calls in real time.
• examples/azure_governed_server.py — the design's Azure-centric reference server.
• examples/redis_quota_server.py — quotas shared across multiple server processes/workers via RedisQuotaStore. Run python examples/redis_quota_server.py --demo to see the limit hold across two simulated workers (uses fakeredis, no real Redis needed for the demo).
• examples/schema_guarded_webhook_server.py — SchemaGuard validating tool arguments against a JSON Schema, with audit records delivered to a webhook via HTTPSink. Run python examples/schema_guarded_webhook_server.py --demo to see a valid call flow through to a real local HTTP collector while invalid ones are rejected before the tool ever runs.

Troubleshooting

HarnessError: ... requires ... Install it with: pip install 'mcp-harness[extra]' — an optional integration (RedisQuotaStore, SchemaGuard, AzureADAuth, harness.run() itself) needs its extra installed. The error names the exact extra; install it and re-run.

An optional integration seems inactive (no spans, approximate token counts, no Prometheus metrics) — run mcp-harness doctor. It lists every extra with an [x]/[ ] marker and prints the interpreter it's running under, which is usually the answer when the extra is installed but into a different environment.

mcp-harness command not found right after pip install mcp-harness — pip warns about this at install time (The script mcp-harness.exe is installed in ... which is not on PATH) but it's easy to miss in a long install log. Either add that Scripts directory to your PATH, or run the CLI as a module instead: python -m mcp_harness.cli daily-rollup ....

Corrupted / unparseable stdio traffic when using StdoutSink — if your server uses the stdio MCP transport (the default), sys.stdout is the protocol channel. Route sinks to sys.stderr (StdoutSink(stream=sys.stderr)), a file (JSONLSink), or a webhook (HTTPSink) instead of the default stdout sink.

A tool call fails with PolicyDenied / QuotaExceeded / AuthenticationError instead of your tool's own exceptions — that's expected: governance middleware rejects a call before your tool runs by raising one of these (all subclasses of HarnessError). Under the mcp SDK these propagate to the client as a normal tool error; in tests, assert on them directly with pytest.raises(...).

Scope

In scope: auth, observability, cost attribution, quotas, audit, policy, resilience, testing.

Out of scope (by design): agent orchestration, a UI/dashboard, a tool registry, and anything that breaks MCP wire-compatibility. A library that does a few things well beats one that does fifteen poorly.

Compatibility & status

• Python 3.10+. Wraps the official mcp SDK (FastMCP) without modifying the wire protocol.
• Beta (v0.1). The Harness, auth, CostTracking, Quotas (including the Redis-backed RedisQuotaStore for multi-process deployments), observability, AllowList, and AuditLog APIs are real and tested. Cloud sinks (Azure Monitor, Event Hubs, Kinesis, Kafka), SchemaGuard, and AzureADAuth hardening are still extension points — base interfaces ship, full wiring is on the roadmap. See CHANGELOG.md.

License

Apache-2.0.