MCP adapter

pip install 'rubric-app[mcp]'

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)

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.