I Built an Open-Source Governance Layer for the Claude API — Here's Why and How

The Anthropic Claude API is great. The SDK is clean, async-native, well-documented. You can go from zero to a working AI feature in an afternoon.

Then someone from your compliance team walks into the room.

"Are we sending PII to a third-party API?"
"Where are the audit logs?"
"How much is this costing per day?"
"How do we control who gets access first?"

The SDK doesn't answer any of these. And it shouldn't — that's not its job.

So I built enterprise-claude-kit: a governance layer that sits between your app and Claude. Here's what it does and how I designed it.

---

Install it

pip install enterprise-claude-kit

That's the whole install. No external services, no Redis, no sidecar containers. SQLite for persistence, pure async Python, zero-infra.

---

The problem in one diagram

Every call goes through five stages:

Your App
  → ① GovernanceLayer (pre-flight) — PII scan, blocked keywords, length guard
  → ② Claude API call
  → ③ GovernanceLayer (post-response) — PII in output, GxP citation check
  → ④ TokenMonitor — cost calculation, budget enforcement, alerts
  → ⑤ AuditLogger — append-only SHA-256 tamper-evident record
  → RunResult { content, cost_usd, tokens, governance_result }

Single-digit millisecond overhead. Everything in-process.

---

Quick start — 3 lines of config

import asyncio
from enterprise_claude import AgentOrchestrator, GovernanceLayer, TokenMonitor

governance = GovernanceLayer(
    pii_filter=True,                          # email, SSN, phone, card, IP
    blocked_keywords=["MNPI", "classified"],
    max_prompt_length=50_000,
)
monitor = TokenMonitor(daily_budget_usd=100.0, alert_threshold_pct=0.80)

async def main():
    async with AgentOrchestrator(governance=governance, monitor=monitor) as orch:
        agent = await orch.create_agent(
            name="Analyst",
            system_prompt="You are a concise financial analyst.",
            persona="research",
        )
        result = await agent.run(
            "Summarise Q3 macro risks in 3 bullets.",
            user_id="[e-mail]"
        )
        print(result.content)
        print(f"${result.cost_usd:.6f}")
        print(result.governance_result.passed)  # True

asyncio.run(main())

---

Module breakdown

🛡️ GovernanceLayer — regex, not AI

I made a deliberate choice: governance decisions use compiled regex patterns, not another model call.

_PII_PATTERNS = {
    "email":       re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'),
    "phone":       re.compile(r'\b(\+1[-.\s]?)?\(?\d{3}\)?[-.\s]\d{3}[-.\s]\d{4}\b'),
    "ssn":         re.compile(r'\b\d{3}-\d{2}-\d{4}\b'),
    "credit_card": re.compile(r'\b(?:\d[ -]?){13,16}\b'),
    "ipv4":        re.compile(r'\b(?:\d{1,3}\.){3}\d{1,3}\b'),
}

Why not spaCy or an NER model? Auditability. A regex is a contract you can read, test, and show a regulator. A neural model is a probability distribution. When compliance asks "how does PII detection work?", you want a one-line answer.

PII detected in a prompt raises GovernanceViolation and aborts — the API is never called. PII in a response is flagged (not raised) — the call already happened.

try:
    result = await agent.run(prompt, user_id="alice")
except GovernanceViolation as e:
    print(e.violation_type)  # "pii_detected" | "blocked_keyword" | "prompt_too_long"

You can also register pre/post hooks — sync or async, detected automatically:

governance = GovernanceLayer(
    pii_filter=True,
    pre_hooks=[lambda prompt, ctx: logger.info(f"Prompt: {prompt[:100]}")],
    post_hooks=[async_verify_response],
)

---

💰 TokenMonitor — budget that survives restarts

Every call is recorded to SQLite:

INSERT INTO usage_records (
    timestamp, agent_id, model, persona,
    input_tokens, output_tokens, cost_usd
) VALUES (?, ?, ?, ?, ?, ?, ?)

Budget state is recalculated from the database — not from in-memory counters. Restart your process; the budget counter is still accurate.

monitor = TokenMonitor(daily_budget_usd=100.0, alert_threshold_pct=0.80)

@monitor.on_alert
async def on_budget_alert(status):
    await slack.post(
        f"⚠️ Budget at {status.pct_used:.0%} — "
        f"${status.used_usd:.2f} of ${status.budget_usd:.2f} used today"
    )

# Query spend
summary = await monitor.get_cost_summary(start_date=yesterday, end_date=now)
print(summary.by_model)    # {"claude-sonnet-4-6": 0.042, "claude-haiku-4-5": 0.008}
print(summary.by_persona)  # {"research": 0.031, "engineering": 0.019}

---

📋 AuditLogger — tamper-evident by design

Three design rules:

1. Append-only — no UPDATE or DELETE anywhere in the codebase
2. SHA-256 checksum on every event (canonical JSON, sort_keys=True)
3. User identity hashed — SHA-256(user_id) stored, never raw email

audit = AuditLogger(db_path="audit.db", gxp_mode=True)

# Verify the entire log — returns {event_id: bool}
checksums = await audit.verify_checksums()
tampered = [eid for eid, ok in checksums.items() if not ok]

if not tampered:
    print(f"✅ All {len(checksums)} events verified clean")

In GxP mode (pharmaceutical 21 CFR Part 11), responses are also checked for [SOURCE:], [REF:], or [CITATION:] markers — the model must cite its sources or the response is flagged.

---

📈 AdoptionTracker — wave rollout with hard gates

wave1 = await tracker.create_wave(name="Architects", target_count=50, order=1)
wave2 = await tracker.create_wave(
    name="Developers", target_count=200, order=2,
    gate_wave_id=wave1.wave_id,
    gate_threshold_pct=0.80,
)

await tracker.activate_wave(wave1.wave_id)

# Try to open Wave 2 too early:
try:
    await tracker.activate_wave(wave2.wave_id)
except WaveGateError as e:
    print(e)
    # "Wave 'Developers' requires 'Architects' to reach 80.0% (currently 12.0%)"

The gate is a WaveGateError, not a process guideline. You can't bypass it.

Literacy scores (0–100) are computed on a log scale from call frequency — log10(calls + 1) / log10(101) * 100. A meaningful proxy for actual engagement vs. click-through adoption.

---

🔌 MCP Registry — connectors without boilerplate

from enterprise_claude import get_connector, list_connectors

github = get_connector("github")
print(github.required_env_vars)  # ["GITHUB_TOKEN"]

for c in list_connectors():
    status = "✅" if c["env_configured"] else "❌ missing env vars"
    print(f"{c['display_name']:<20} {status}")

10 pre-built connectors: GitHub, Jira, Slack, Confluence, SharePoint, PostgreSQL, ServiceNow, Salesforce, Teams, local filesystem.

---

⌨️ CLI — operations without code

ecl cost summary --days 7        # spend breakdown by model and persona
ecl audit query --persona research --result pass
ecl audit export --format csv --out trail.csv
ecl waves list                   # all waves + progress bars
ecl waves activate <wave-id>
ecl connectors list              # env-var status for all 10 connectors
ecl connectors validate github

---

Design decisions worth explaining

Why no LangChain?

LangChain is a framework for chaining LLM calls. This library is a governance layer for production deployments. Those are different problems. Adding LangChain would mean:

• 50+ transitive dependencies
• Abstractions that obscure the direct API call
• A framework version pinned to your governance layer version

This library has one job: make every Claude call governed, tracked, and audited. It calls anthropic.AsyncAnthropic.messages.create() directly. No chains, no agents framework, no vector stores.

Why SQLite and not PostgreSQL?

SQLite is zero-infra. You can start using this library without a database server. The schema is simple enough that a ATTACH DATABASE migration to PostgreSQL with asyncpg is documented if you need it for production scale. Start simple, scale when you have the problem.

Why frozen Pydantic configs?

config = GovernanceConfig(pii_filter=True, blocked_keywords=["MNPI"])
config.pii_filter = False  # raises ValidationError

Governance rules must not drift during a running process. An immutable config is an enforcement mechanism, not a style preference.

---

What's in the repo

enterprise_claude/
├── governance.py        # GovernanceLayer — the policy engine
├── orchestrator.py      # AgentOrchestrator — agent lifecycle
├── token_monitor.py     # TokenMonitor — cost accounting
├── audit.py             # AuditLogger — immutable event log
├── adoption_tracker.py  # AdoptionTracker — wave rollout
├── mcp_connectors.py    # MCPConnectorRegistry — connector catalogue
└── cli.py               # ecl command — management CLI

tests/                   # 66 pytest tests
examples/
├── basic_governed_agent.py    # ← start here
├── clinical_trial_agent.py    # GxP + audit trail
└── sdlc_accelerator.py        # wave simulation (no API key needed)

---

Try it

pip install enterprise-claude-kit
cp .env.example .env     # add ANTHROPIC_API_KEY
python examples/basic_governed_agent.py

GitHub: github.com/sairajboddula/enterprise-claude-kit
PyPI: pypi.org/project/enterprise-claude-kit

MIT licensed. PRs welcome. If you're building Claude integrations in a regulated industry or large enterprise, I'd love to hear what you're building.

---

What governance problems are you solving in your Claude deployments? Drop a comment below.