MCP Enterprise Tool Gateway: A Governed Model Context Protocol Server for Bank-Grade Agent Tool Access

Fictional bank notice: "Northbridge Financial Group" is a wholly invented bank, created solely as a synthetic demo brand for this project. It does not exist. Every account, customer, transaction, and general ledger entry referenced anywhere in this repository is fabricated for illustration -- no real institution, customer, or employer is represented here.

Why this exists

A regulated bank cannot let every agent framework its teams adopt -- LangGraph here, CrewAI there, an in-house AutoGen pilot somewhere else -- reimplement its own access control, PII handling, and audit logging around "who gets to call the account-lookup tool." That inconsistency is where incidents come from. This repository shows the alternative: one governed MCP server that every agent framework talks to identically, where RBAC, PII redaction, rate limiting, and append-only audit logging are enforced exactly once, in the tool-serving layer itself, instead of trusted to each calling agent's own good behavior. It is deliberately built as zero-paid-API-key infrastructure -- this repo serves tools, it does not call an LLM -- so a reviewer can clone it and see the whole governance pipeline run locally in seconds.

Related MCP server: production-grade-mcp-agentic-system

Architecture

flowchart LR
    subgraph Callers["Agent Frameworks (examples/)"]
        LG["LangGraph<br/>ToolNode"]
        CA["CrewAI<br/>BaseTool"]
        GC["Generic MCP client<br/>(any framework)"]
    end

    subgraph Gateway["MCP Enterprise Tool Gateway"]
        MCP["mcp_server.py<br/>FastMCP tool handlers"]
        MW["ToolGateway.invoke (gateway.py)<br/>governance pipeline"]
        RBAC["RBAC scope check<br/>(rbac.py)"]
        RL["Rate limiter<br/>(ratelimit.py)"]
        RED["PII redaction<br/>(redaction.py)"]
        AUD["Append-only audit log<br/>(audit.py, SQLite)"]
        CB["Circuit breaker<br/>(resilience.py)"]
    end

    subgraph Backend["Mock Enterprise Systems"]
        DS["MockDatastore (store.py)<br/>accounts / transactions / KYC /<br/>product catalog / general ledger"]
    end

    subgraph Ops["Admin / Audit API (admin_api.py)"]
        H["/healthz /readyz"]
        M["/metrics (Prometheus)"]
        A["/audit /scopes /traces"]
    end

    LG -->|MCP: stdio / streamable-http| MCP
    CA -->|MCP: stdio / streamable-http| MCP
    GC -->|MCP: stdio / streamable-http| MCP

    MCP --> MW
    MW --> RBAC
    RBAC -->|allowed| RL
    RL -->|under limit| CB
    CB --> DS
    DS --> RED
    RED --> AUD
    AUD -->|redacted result| MCP

    RBAC -.->|denied| AUD
    RL -.->|rate limited| AUD

    Ops -.->|reads| AUD
    Ops -.->|reads| RBAC

Every arrow into "Gateway" is the same regardless of which box in "Callers" it came from -- that uniformity is the entire point. See GOVERNANCE.md for the detailed per-control walkthrough.

Key Design Decisions

1. agent_identity is an explicit tool argument, not transport metadata. MCP's stdio transport -- the most portable, zero-install option, and what every "generic MCP client" example uses -- has no standard out-of-band header mechanism the way HTTP does. Rather than build governance on a feature only some transports support, every tool accepts agent_identity directly (see models.AgentContext). Tradeoff: the identity is currently self-asserted by the caller. In a production deployment behind streamable-HTTP, this string would be replaced, not merely supplemented, by an identity derived from mTLS or a verified OAuth token at the transport boundary.

2. SQLite for the audit log and RBAC table, not Postgres. This is a demo/portfolio project: SQLite requires no separate service to run, ships in the standard library, and is genuinely sufficient for an append-mostly, single-writer audit log. Tradeoff: a single SQLite file does not support multiple concurrent writer processes across nodes, which is why the Kubernetes manifests intentionally ship at 1 replica per component (see the comment in deploy/k8s/deployment.yaml). Migrating sqlite_backend.py to Postgres is a self-contained, scoped change precisely because storage access is centralized in one module.

3. Redaction is uniform and pattern-based, with no per-field allowlist. Every string matching an SSN or account-number-shaped digit run is masked to its last 4 characters -- including primary-key fields like account_id in the response for the very account the caller asked about. Tradeoff: this occasionally masks a value that arguably "should" be visible (mild usability cost), in exchange for never requiring a human to maintain a growing "these fields are safe" exception list as new tools are added, which is exactly the kind of list that quietly rots.

4. In-memory rate limiting, not a shared store. The token-bucket limiter in ratelimit.py is process-local state. Tradeoff: a horizontally scaled deployment would need a shared backend (Redis, most likely) for limits to be enforced consistently across replicas; that is flagged explicitly in the module rather than silently glossed over.

5. A real (if simple) circuit breaker around a datastore that cannot actually fail. The mock datastore is an in-memory Python dict -- it cannot time out or 500. resilience.py wraps it in a genuine three-state circuit breaker anyway, because the pattern a production version of this gateway needs (fail fast and stop hammering a struggling downstream system during an incident) should exist and be tested now, not bolted on later under pressure. /readyz reports the breaker's state, which is what a Kubernetes readiness probe should key off in production.

Governance & Guardrails

This is the centerpiece of the repo -- see GOVERNANCE.md for the full walkthrough. In summary, every tool call passes through the same five controls, enforced once in ToolGateway.invoke regardless of caller:

1. Identity -- every call carries an explicit agent_identity.

2. RBAC (rbac.py) -- default-deny scope table, SQLite-backed. A customer-service-agent identity can call lookup_account_summary but is denied get_general_ledger_entry; see tests/test_rbac.py.

3. Rate limiting (ratelimit.py) -- per-identity token bucket; trips raise a clear MCP tool error with a retry hint.

4. PII redaction (redaction.py) -- SSN/account-number-shaped values masked to their last 4 characters in every response and before anything is written to the audit log.

5. Append-only audit log (audit.py) -- one row per call (including denials/failures) with redacted params, a result hash, latency, and status. No update/delete code path exists.

This gateway is the enforcement point that keeps raw sensitive data out of an LLM's context window, no matter which agent framework is calling it. Redaction happens inside the gateway, before a result is ever serialized back over MCP -- an agent framework cannot opt out of it, and does not need to know it is happening.

Getting Started

Requires Python 3.11+. No paid API keys, no external services -- the gateway serves tools backed by an in-memory mock datastore.

git clone <this-repo-url>
cd mcp-enterprise-tool-gateway
python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate

make install                     # pip install -e . plus dev tooling
cp .env.example .env             # optional -- sane defaults work as-is

make test                        # 39 tests: RBAC, audit, redaction, rate limiting, ...
make lint                        # ruff + mypy

Run the MCP server

make run                         # python -m mcp_gateway (stdio transport by default)

Point a generic MCP client at it

examples/generic_mcp_client.py is genuinely runnable end-to-end -- it spawns the server over stdio, lists the five governed tools, calls lookup_account_summary as customer-service-agent (showing PII redaction in the response), then calls get_general_ledger_entry with the same identity to show an RBAC denial surfacing as a clean MCP tool error:

python examples/generic_mcp_client.py

To point any other MCP-compatible client (Claude Desktop, mcp dev, a hand-rolled ClientSession, LangGraph's langchain-mcp-adapters, ...) at this server, the shape is the same in every case: spawn python -m mcp_gateway as a stdio subprocess (or connect to GATEWAY_MCP_TRANSPORT=streamable-http over HTTP once deployed), then call any of the five tools with an agent_identity string plus that tool's declared parameters. See examples/README.md for the LangGraph and CrewAI integration sketches.

Run the admin/audit API

make run-admin-api               # http://localhost:8000
curl localhost:8000/healthz
curl localhost:8000/readyz
curl localhost:8000/scopes       # current RBAC table
curl localhost:8000/audit        # recent audit log entries (PII-redacted)
curl localhost:8000/metrics      # Prometheus exposition format

One-command local stack

docker compose up --build        # admin API on :8000, MCP server (streamable-http) on :8765

Observability

Four Prometheus series (mcp_gateway/metrics.py), scraped from /metrics on the admin API, are the headline governance signals for a Grafana dashboard:

• mcp_gateway_tool_calls_total{tool_name, agent_identity, status} -- call volume broken down by who is calling what, and whether it succeeded, was denied, rate-limited, timed out, or errored.

• mcp_gateway_redaction_hits_total{tool_name} -- how often PII redaction actually fires, per tool.

• mcp_gateway_rbac_denials_total{tool_name, agent_identity} -- RBAC denial rate, per identity.

• mcp_gateway_rate_limit_trips_total{agent_identity} -- rate-limit trip rate, per identity.

• mcp_gateway_tool_call_latency_seconds{tool_name} -- end-to-end latency histogram, including governance overhead.

Every call also gets a lightweight tracing span (tracing.py) with a correlation ID shared with its structured JSON log line and audit row, and the most recent 200 spans are queryable at /traces for local debugging. In production, these spans are structured enough to ship into an OpenTelemetry collector / Jaeger without changing any call sites; wire Prometheus to scrape /metrics and build a Grafana dashboard on the four series above as the natural next step (see Roadmap).

Production Deployment

• Docker: Dockerfile is a multi-stage build (builder stage installs pinned dependencies into a venv; runtime stage is a slim, non-compiler-bearing image) running as a non-root gateway user (uid

  10001. on a pinned python:3.12.8-slim-bookworm base.

• docker-compose: docker-compose.yml brings up the admin API and the MCP server (streamable-HTTP transport) as two containers from one image, sharing a named volume for the SQLite-backed audit log.

• Kubernetes: deploy/k8s/ has a Deployment + Service + PVC per component, a ConfigMap for the (all non-secret) environment variables, a HorizontalPodAutoscaler for the stateless admin API, and a default-deny NetworkPolicy with explicit allow rules. See the comments in deployment.yaml for why replica count is intentionally 1 until the Postgres migration described above lands.

• OpenShift: deploy/OPENSHIFT.md covers SCC/UID handling, Routes vs. Ingress, ImageStreams, and NetworkPolicy support on OpenShift's default SDN.

• CI: .github/workflows/ci.yml runs ruff, mypy, pytest, and a Docker build on every push/PR, in that order, failing fast on the cheapest check first.

Tech Stack

Repository Structure

mcp-enterprise-tool-gateway/
├── src/mcp_gateway/
│   ├── mcp_server.py          # MCP protocol adapter (FastMCP tool handlers)
│   ├── gateway.py             # ToolGateway.invoke -- the governance pipeline
│   ├── container.py           # Wires config -> storage -> rbac -> gateway
│   ├── config.py              # pydantic-settings, GATEWAY_* env vars
│   ├── models.py              # Pydantic models for every boundary
│   ├── rbac.py                # Default-deny scope table (SQLite-backed)
│   ├── ratelimit.py           # Per-agent-identity token bucket
│   ├── redaction.py           # PII masking middleware
│   ├── audit.py               # Append-only audit log (SQLite-backed)
│   ├── resilience.py          # Circuit breaker around the mock datastore
│   ├── tracing.py             # Lightweight per-call tracing spans
│   ├── metrics.py             # Prometheus counters/histograms
│   ├── logging_setup.py       # Structured JSON logging + correlation IDs
│   ├── correlation.py         # contextvars-based correlation ID propagation
│   ├── sqlite_backend.py      # Shared SQLite connection/schema plumbing
│   ├── admin_api.py           # FastAPI: /healthz /readyz /metrics /audit /scopes /traces
│   ├── datastore/
│   │   ├── seed_data.py       # Synthetic Northbridge Financial Group fixtures
│   │   └── store.py           # In-memory mock enterprise datastore
│   └── tools/
│       └── banking_tools.py   # The 5 tool implementations
├── tests/                     # RBAC, audit, redaction, rate limiting, gateway, admin API
├── examples/
│   ├── generic_mcp_client.py  # Actually runnable end-to-end
│   ├── langgraph_tool_node.py # Illustrative LangGraph integration sketch
│   └── crewai_custom_tool.py  # Illustrative CrewAI integration sketch
├── deploy/
│   ├── k8s/                   # Deployment, Service, ConfigMap, HPA, NetworkPolicy, PVC
│   └── OPENSHIFT.md
├── .github/workflows/ci.yml   # lint -> typecheck -> test -> docker build
├── Dockerfile                 # Multi-stage, non-root, pinned base image
├── docker-compose.yml
├── Makefile
├── GOVERNANCE.md              # Governance & guardrails deep dive
├── SECURITY.md
└── CONTRIBUTING.md

One governed tool surface, many agent frameworks

This gateway is designed to be consumed identically from LangGraph, CrewAI, AutoGen, or any other MCP-compatible client -- examples/ proves the shape for three of them. None of them get special treatment, and none of them can bypass RBAC, redaction, rate limiting, or audit logging, because those controls live in the gateway, not in any particular framework's integration code. That is the connective tissue across a larger personal portfolio of agent-framework demos: whichever framework a given repo showcases, the tool access layer underneath it is the same governed pattern shown here. (This repo stands alone -- it does not assume or require any of those other demos to be present or wired in.)

Roadmap / What I'd Build Next

• Postgres-backed audit log to remove the single-writer/single-replica constraint noted in sqlite_backend.py and deploy/k8s/deployment.yaml, enabling true horizontal scaling of the MCP server.

• Verified identity at the transport layer -- replace the self-asserted agent_identity argument with an identity derived from mTLS client certificates or a validated OAuth token, terminated at the streamable-HTTP boundary, so RBAC decisions are keyed off something a caller cannot spoof.

• OpenTelemetry exporter wired directly into tracing.py's spans, plus a checked-in Grafana dashboard JSON built on the four Prometheus series already exposed.

• Shared (Redis-backed) rate limiter so limits hold consistently once the MCP server runs as more than one replica.

• Dependabot / SCA scanning wired into CI, now that dependencies are pinned to exact versions.

• A real langchain-mcp-adapters-based LangGraph integration test, run in CI against this server, once that dependency's install footprint is acceptable for a lightweight infra repo.