Mcp Governed Access

mcp-governed-access

Governed MCP server access with RBAC, append-only audit logging, and circuit breaking.

Repository: github.com/rationallycreative/mcp-governed-access

Python 3.11+. MIT license.

---

This framework was built after three years of shipping production AI in healthcare environments where governance failures have real patient consequences. The RBAC pattern came from HIPAA. The audit log design came from compliance reviews. The circuit breaker came from watching cloud provider outages take down production pipelines at 2am. These are not theoretical principles.

It is now used across healthcare, fintech, legal, and HR analytics examples in this repository — the same three primitives applied to different regulatory contexts.

---

Who This Is For

If you are building AI agents that touch clinical records, financial systems, regulated data, internal knowledge bases, or any environment where an auditor might someday ask "what did the AI access and when" — this framework is for you.

If cloud APIs are not an option because your data cannot leave the environment, the sovereign deployment mode runs entirely on local inference (Ollama), local embeddings, and local vector storage. No API keys required.

If you are building a demo chatbot with public data, you probably do not need this.

For the reasoning behind every architectural decision, see [Governance Philosophy](docs/governance-philosophy.md).

---

The Problem

Enterprise AI agents call MCP servers the same way they call any other tool: repeatedly, autonomously, and at scale. That is fine when the server exposes public data. It is not fine when the server reaches into clinical records, financial systems, or internal knowledge bases.

Most teams bolt access control on after the fact. They filter tool output in the agent layer. They hide sensitive fields in post-processing. They trust the LLM not to ask for the wrong thing. This fails in predictable ways:

Filtering output is not scoping access. A role that can call get_document but receives a redacted response still proved the document exists. A role that can see ingest_document in the tool list knows ingestion is possible, even if the call is denied at runtime. Visibility and permission must be the same gate, decided at registration time, not patched after execution.

An audit log that can be modified is not an audit log. Compliance reviewers and incident investigators need to reconstruct what happened without trusting the actors who caused it. If records can be updated or deleted, the log is a narrative, not evidence. Append-only storage is the minimum credible contract.

Downstream APIs fail at 2am. Rate limits flip from 200 to 503 in a minute. DNS blips last long enough to exhaust retry budgets. An agent that keeps hammering a dead endpoint wedges the whole pipeline. A circuit breaker stops making things worse: fast-fail when the upstream is down, one careful probe when recovery is possible, full traffic only after the upstream earns it back.

This project bakes governance into the execution path from the start. If a tool runs, it ran through RBAC, circuit breaking, and audit. If it did not run, that denial is in the audit log.

---

The Three Primitives

1. RBAC at the tool boundary

Tools are registered at server startup with an explicit role set. get_visible_tools(role) returns exactly the tools that role may invoke. There is no hidden-but-callable path. A reader on the knowledge server sees search_documents only. get_document and ingest_document do not appear in discovery. They are invisible, as if they do not exist.

Denied calls raise PermissionError after the attempt is logged as blocked.

2. Append-only audit logging

Every governed invocation appends one JSON line: timestamp, role, server, tool, sanitized parameters, response status, latency, request ID. There is no update or delete API. AuditLogger.log() never raises to the caller. Tool traffic must not fail because disk is full. You monitor write health separately. Silent audit loss is still a governance failure.

3. Circuit breaker

Async circuit breaker with CLOSED, OPEN, and HALF_OPEN states. Configurable failure threshold, recovery timeout, and half-open probe capacity. Built from real provider outage patterns: when the upstream is dead, stop retrying and return a clear fast-fail so callers can degrade gracefully.

---

Architecture

core/           Governance primitives (audit, RBAC, circuit breaker)
servers/        Domain MCP servers inheriting GovernedMCPServer
eval/           Golden eval harness and JSONL case files
examples/       Usage examples (scaffold)
tests/          Pytest suite exercising real governance paths

core/ Reusable governance layer. No domain logic. AuditLogger / AuditReader, RBACRegistry, CircuitBreaker / CircuitBreakerRegistry.

servers/ Domain servers. base_server.py defines GovernedMCPServer and the mandatory execute_tool() path. knowledge/ is the reference implementation: RAG search, document fetch, and ingestion with reader/analyst/admin roles.

eval/ GoldenEvalHarness runs structured JSONL cases against any GovernedMCPServer. Cases assert RBAC, audit, validation, circuit breaking, and domain behavior. Results export to JSONL for audit trail.

examples/ Reference implementations across regulated industries: healthcare (knowledge server), fintech (transaction, account, loan, compliance servers), legal (matter, document, communication, conflict servers), and HR analytics (candidate, employee, compliance, protected data servers). Each example includes a role visibility table, a demo script, and a README explaining the regulatory context.

Execution path

Every tool call flows through GovernedMCPServer.execute_tool():

1. RBAC check. Denied calls log blocked and raise PermissionError.
2. Circuit breaker wrap around handler execution.
3. Audit log for success, error, or blocked.
4. Structured ToolResult returned to the caller.

Handlers live in a private _tools dict. They are not callable except through this path.

---

Quick Start

Install

git clone https://github.com/rationallycreative/mcp-governed-access.git
cd mcp-governed-access
pip install -e ".[dev]"

Or use the Makefile (creates a venv automatically):

make install
make test    # 46 tests
make lint

Environment variables

| Variable | Default | Description | |----------|---------|-------------| | AUDIT_LOG_PATH | ./audit.log | Append-only audit log file path |

Run the knowledge server

make run

Output shows the visibility difference across roles:

KnowledgeServer ready (knowledge)
  admin: get_document, ingest_document, search_documents
  analyst: get_document, search_documents
  reader: search_documents

Same server. Same tools registered. Different advertised surface per role.

Exercise reader vs admin in Python

import asyncio
from servers.knowledge.server import create_knowledge_server

server = create_knowledge_server()

# Reader: search only
reader_tools = [t["name"] for t in server.list_visible_tools("reader")]
print(reader_tools)  # ['search_documents']

# Admin: all three tools visible
admin_tools = [t["name"] for t in server.list_visible_tools("admin")]
print(admin_tools)  # ['get_document', 'ingest_document', 'search_documents']

async def demo() -> None:
    # Admin ingests a document
    ingest = await server.execute_tool(
        "admin",
        "ingest_document",
        {
            "content": "Hand hygiene is mandatory in all clinical zones.",
            "metadata": {"source": "infection-control"},
            "document_id": "DOC-001",
        },
    )
    print(ingest.status, ingest.result)

    # Reader can search
    search = await server.execute_tool(
        "reader",
        "search_documents",
        {"query": "hand hygiene", "top_k": 3},
    )
    print(search.status, len(search.result))

    # Reader cannot fetch full documents
    try:
        await server.execute_tool("reader", "get_document", {"document_id": "DOC-001"})
    except PermissionError as exc:
        print("denied:", exc)

asyncio.run(demo())

Every call above is logged to AUDIT_LOG_PATH. Denied calls included.

Run the golden eval harness

from eval.harness import GoldenEvalHarness
from servers.knowledge.server import create_knowledge_server

server = create_knowledge_server()
harness = GoldenEvalHarness()
result = harness.run_from_file(server, "eval/cases/knowledge_cases.jsonl")
print(f"{result.passed}/{result.total} passed")
harness.export_results(result, "eval/results.jsonl")

---

Dashboard

A React dashboard ships with the project for interactive exploration of the governance layer.

Screenshot placeholder: role visibility matrix showing reader vs admin surface

Start the full development environment:

make dev

Open http://localhost:5173

The dashboard demonstrates:

• Role-scoped tool visibility — switch roles and watch the available tools change
• Live audit log — every tool call, denial, and circuit breaker event appears in real time
• Role visibility matrix — see every role's surface across every server at a glance
• Circuit breaker status — trip and recover breakers to demonstrate degradation behavior

The dashboard is a development and demonstration tool. It is not a production admin interface.

See [dashboard/README.md](dashboard/README.md) for component details, screenshot guide, and a suggested demo walkthrough.

---

Building Your Own Domain Server

Subclass GovernedMCPServer and implement register_tools(). Five steps:

1. Declare your deployment roles.

from core.rbac import DeploymentRoleConfig, RBACRegistry

MY_ROLES = frozenset({"viewer", "operator", "admin"})
rbac = RBACRegistry(DeploymentRoleConfig(roles=MY_ROLES))

2. Wire governance dependencies.

from pathlib import Path
from core.audit import AuditLogger
from core.circuit_breaker import CircuitBreakerConfig, CircuitBreakerRegistry
from servers.base_server import GovernedMCPServer, GovernedTool

audit = AuditLogger(path=Path("./audit.log"))
breakers = CircuitBreakerRegistry()
breakers.register("my-server", CircuitBreakerConfig())

3. Subclass and register tools in register_tools().

class MyServer(GovernedMCPServer):
    def register_tools(self) -> None:
        self._add_tool(
            GovernedTool(
                name="list_items",
                description="List items visible to the caller.",
                parameter_schema={"type": "object", "properties": {}},
                handler=self._list_items,
            ),
            allowed_roles=["viewer", "operator", "admin"],
        )
        self._add_tool(
            GovernedTool(
                name="delete_item",
                description="Delete an item by ID.",
                parameter_schema={
                    "type": "object",
                    "properties": {"item_id": {"type": "string"}},
                    "required": ["item_id"],
                },
                handler=self._delete_item,
            ),
            allowed_roles=["admin"],
        )

    async def _list_items(self, parameters: dict) -> list[dict]:
        return [{"id": "1", "name": "example"}]

    async def _delete_item(self, parameters: dict) -> dict:
        return {"deleted": parameters["item_id"]}

4. Construct and start.

server = MyServer(
    server_name="my-server",
    rbac_registry=rbac,
    audit_logger=audit,
    circuit_breakers=breakers,
)
server.startup()

5. Route all traffic through execute_tool().

result = await server.execute_tool("viewer", "list_items", {})
visible = server.list_visible_tools("viewer")

Never call handlers directly. Never register tools outside _add_tool(). MCP wiring should call execute_tool() and list_visible_tools() only.

---

Industry Examples

The examples/ directory contains reference implementations for four regulated industries. Each demonstrates the same three governance primitives applied to a different regulatory context.

| Industry | Regulations | Example Servers | Key Governance Pattern | |----------|-------------|-----------------|------------------------| | Healthcare | HIPAA, HITECH | knowledge | PHI-aware retrieval scoping, clinical severity bypass | | Fintech | PCI DSS, GLBA, SEC, ECOA | transaction, account, loan, compliance | Customer ID hashing, adverse action audit trail | | Legal | ABA Rule 1.6, Work Product Doctrine | matter, document, communication, conflict | Matter-scoped isolation, privilege level on every retrieval | | HR Analytics | EEOC, NYC LL144, GDPR Art. 9 | candidate, employee, compliance, protected_data | Schema-level protected class exclusion, k-anonymity enforcement |

Run any industry demo:

python examples/fintech/fintech_demo.py
python examples/legal/legal_demo.py
python examples/hr_analytics/hr_analytics_demo.py

Each demo prints a role visibility table, executes cross-role scenarios, demonstrates privilege escalation blocking, and prints the full audit log with regulation tags.

---

The KnowledgeStore Protocol

The knowledge server does not care which vector backend is active. KnowledgeStore is a typing.Protocol with three methods: search(), get(), and ingest(). The governance layer depends on the protocol, not on ChromaDB, Pinecone, or an in-memory cosine similarity hack.

The default InMemoryKnowledgeStore is pure Python. Good for tests and local demos. ChromaKnowledgeStore provides persistent ChromaDB storage with local sentence-transformers embeddings. Pass a chroma_host URL to use the Chroma HTTP service in Docker:

from servers.knowledge.server import KnowledgeServer
from servers.knowledge.chroma_store import ChromaKnowledgeStore

store = ChromaKnowledgeStore("./chroma_db", chroma_host="http://chromadb:8000")
server = KnowledgeServer(store=store, rbac_registry=rbac, audit_logger=audit, circuit_breakers=breakers)
server.startup()

RBAC and audit behavior stay identical regardless of backend.

---

Eval Harness

The eval harness is not a disposable test suite. It is a governance artifact. Every passing case in eval/cases/knowledge_cases.jsonl is a permanent assertion about system behavior: RBAC denials, visibility rules, validation errors, circuit breaker fast-fail, audit log growth on denied calls, self-citation deprioritization.

When a failure mode appears in production, you add a new case. You extend categories of behavior. You do not patch individual incidents and move on. GoldenEvalHarness runs against any GovernedMCPServer, so golden cases travel with the governance contract, not one server's internals.

---

Design Decisions

Why direct API over frameworks

GovernedMCPServer is a small abstract base class, not a framework with plugins, middleware stacks, and configuration DSLs. You subclass it, register tools, and call execute_tool(). The execution path fits in one screen. You can explain it in a security review without a diagram of interceptors. Frameworks hide governance behind conventions. When governance is the product, hiding it is a liability.

Why RBAC at access time, not post-filter

Post-execution filtering leaks existence. A forbidden tool can still appear in MCP discovery. A denied call still proves the resource is there. Registration-time RBAC binds each tool to an explicit role set at startup. get_visible_tools() and is_allowed() read the same binding. The advertised surface equals the authorized surface. No drift between what the server shows and what it permits.

Why append-only

Investigators need to trust the log more than they trust the operators. Append-only JSONL is simple

…

Source & license

This open-source MCP server is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: Rationally-Creative
• Source: Rationally-Creative/mcp-governed-access
• License: MIT
• Homepage: https://wholemindedbuilder.substack.com

Install and usage instructions live in the source repository linked above.