# 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.

## Architecture

```mermaid
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](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](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.

```bash
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

```bash
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:

```bash
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

```bash
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

```bash
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

| Layer | Technology | Why |
|---|---|---|
| MCP protocol | [`mcp`](https://pypi.org/project/mcp/) (official Python SDK), `FastMCP` | The standard this whole portfolio's tool-access layer is built on. |
| Admin/audit HTTP API | FastAPI + Uvicorn | Async, typed, first-class OpenAPI docs, minimal boilerplate for health/metrics/audit endpoints. |
| Validation | Pydantic v2 | Boundary validation on every tool input/output and audit record. |
| Configuration | pydantic-settings | Typed, validated env-var configuration; no hardcoded secrets. |
| Audit log / RBAC storage | SQLite (stdlib `sqlite3`) | Zero-install, sufficient for an append-mostly audit log; see Key Design Decisions. |
| Metrics | `prometheus-client` | Standard `/metrics` exposition format for Grafana dashboards. |
| Logging | `python-json-logger` | Structured JSON logs with correlation IDs, safe for a real log aggregator. |
| Testing | pytest, pytest-asyncio, httpx | 39 unit/integration tests covering every governance control. |
| Lint / types | ruff, mypy | Fast style+correctness checks and static typing, enforced in CI. |
| Containerization | Docker (multi-stage, non-root), docker-compose | One-command local spin-up; production-shaped image. |
| Orchestration | Kubernetes manifests, OpenShift notes | Deployment/Service/PVC/HPA/NetworkPolicy for a real cluster. |

## 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.


  ---


### Thank you for reading

#### Please consider giving a star if you find the repo useful. Thank you.

---

### **AUTHOR'S BACKGROUND**
### Author's Name:  Emmanuel Oyekanlu
```
Skillset:   I have experience spanning several years in data science, enterprise AI architecture and solutions, developing scalable enterprise data pipelines,
enterprise solution architecture, architecting enterprise systems data and AI applications,
software and AI solution design and deployments, data engineering, industrial intelligent vision systems, high performance computing (GPU, CUDA), machine learning,
NLP, Agentic-AI and LLM applications as well as deploying scalable solutions (apps) on-prem and in the cloud.

I can be reached through: manuelbomi@yahoo.com

Publications:  https://scholar.google.com/citations?user=S-jTMfkAAAAJ&hl=en
LinkedIn:  https://www.linkedin.com/in/emmanuel-oyekanlu-6ba98616
Github:  https://github.com/manuelbomi

```
[![Icons](https://skillicons.dev/icons?i=aws,azure,gcp,scala,mongodb,redis,cassandra,kafka,anaconda,matlab,nodejs,django,py,c,anaconda,git,github,mysql,docker,kubernetes&theme=dark)](https://skillicons.dev)