我为 Claude API 构建了一个开源治理层：原因与实现

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 Enter fullscreen mode Exit fullscreen mode 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 } Enter fullscreen mode Exit fullscreen mode 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="alice@acme.com" ) print(result.content) print(f"${result.cost_usd:.6f}") print(result.governance_result.passed) # True asyncio.run(main()) Enter fullscreen mode Exit fullscreen mode 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'), } Enter fullscreen mode Exit fullscreen mode 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" Enter fullscreen mode Exit fullscreen mode 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], ) Enter fullscreen mode Exit fullscreen mode 💰 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 (?, ?, ?, ?, ?, ?, ?) Enter fullscreen mode Exit fullscreen mode 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} Enter fullscreen mode Exit fullscreen mode 📋 AuditLogger — tamper-evident by design Three design rules: Append-only — no UPDATE or DELETE anywhere in the codebase SHA-256 checksum on every event (canonical JSON, sort_keys=True) 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") Enter fullscreen mode Exit fullscreen mode 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%)" Enter fullscreen mode Exit fullscreen mode 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']: ecl connectors list # env-var status for all 10 connectors ecl connectors validate github Enter fullscreen mode Exit fullscreen mode 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 Enter fullscreen mode Exit fullscreen mode 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) Enter fullscreen mode Exit fullscreen mode Try it pip install enterprise-claude-kit cp .env.example .env # add ANTHROPIC_API_KEY python examples/basic_governed_agent.py Enter fullscreen mode Exit fullscreen mode 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.