> ## Documentation Index
> Fetch the complete documentation index at: https://docs.rubric-app.com/llms.txt
> Use this file to discover all available pages before exploring further.

# MCP adapter

> Govern any MCP-speaking agent with one wrapper call.

The MCP adapter wraps an `mcp.ClientSession` so every `call_tool()` is gated by the governance evaluator and produces an audit event. Denied calls come back as a normal `CallToolResult(isError=True)` — your harness doesn't need any special-casing.

## Install

```bash theme={null}
pip install 'rubric-app[mcp]'
```

The `[mcp]` extra pulls `mcp` as a dependency. If you already have it installed, `pip install rubric-app` is enough.

## Basic usage

```python theme={null}
from rubric import Governance
from rubric.adapters.mcp import govern_mcp_session
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

server = StdioServerParameters(command="python", args=["my_mcp_server.py"])

with Governance.bootstrap(agent_name="ops-bot") as gov:
    async with stdio_client(server) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            wrapped = govern_mcp_session(gov, session, session_id="ops-session")

            # Same MCP API — just call methods on `wrapped` instead of `session`.
            tools = await wrapped.list_tools()
            result = await wrapped.call_tool("list_files", {"path": "/tmp"})

            if result.isError:
                # The model sees this and adapts.
                print(result.content[0].text)
```

The wrapper exposes the same methods as `ClientSession` — `list_tools`, `call_tool`, etc. — so it's a drop-in replacement.

## What gets evaluated

Every `wrapped.call_tool(name, args)` triggers `gov.evaluate(name, session_id=…, metadata=EvaluationMetadata(input=args))`. The `tool_name` field in your policies matches the MCP tool name as the server advertised it.

* **`tool_name`** = the MCP tool name (e.g. `list_files`, `delete_file`).
* **`input.<key>`** = entries in the `args` dict (e.g. `input.path`).
* **`framework`** = `"mcp"` on the audit event.

## Denied calls

When a policy denies the call:

1. `evaluate()` returns `decision="deny"` with the rule's reason.
2. The adapter constructs a `CallToolResult(isError=True, content=[TextContent(...)])` containing the deny reason.
3. The actual MCP server is **never called**.
4. The audit event ships with `decision="deny"` and the policy id.

The model sees the error message in the tool result and reacts — usually by trying a different approach or asking the user. No special harness logic required.

## With traces

Pass a `TraceContext` and it's used for every governed call in this session:

```python theme={null}
from rubric import TraceContext, UserMessage

trace = TraceContext()
trace.append(UserMessage(content="Clean up old logs"))

wrapped = govern_mcp_session(gov, session, session_id="ops-session", trace=trace)

await wrapped.call_tool("list_files", {"path": "/var/log"})
await wrapped.call_tool("delete_file", {"path": "/var/log/old.log"})
# Both calls are recorded in `trace`, including the assistant's tool_use
# messages and the tool_result responses.
```

## Custom session id strategy

`session_id` is whatever string makes sense for your harness. Two common choices:

* **Per-conversation.** Use the user's chat thread id — every audit event for that conversation groups under one id.
* **Per-process.** Use `os.getpid()` or a uuid generated at startup — useful for batch jobs.

The audit log groups by session id, so picking the right granularity makes triaging much faster.

## Full example

A working end-to-end script ships with the SDK at `examples/mcp_quickstart.py`. It spawns an in-tree FastMCP server, governs the client, and runs an `allow` and a `deny` call against the bundle.