MCP Series (08): Enterprise Governance — Registry, Routing, and Observability

When an enterprise has 20+ MCP Servers, three governance problems emerge simultaneously: how to discover the right Server (Registry), how to route tool calls to the correct Server, and how to know when a tool call fails (observability). This article covers complete engineering solutions for all three, including Registry YAML design, Langfuse integration for MCP call chains, and alert rule templates.

·7 min read·AI Engineering
MCPGovernanceRegistryObservabilityLangfuse

Why Governance Becomes Necessary

Three MCP Servers are manageable from memory. Twenty require a system.

As the number of MCP Servers in an enterprise grows, predictable problems emerge:

  • A new engineer builds a Jira tool that already exists because there's no directory
  • An Agent calls the deprecated search_jira tool (v1.x) instead of the current search_issues (v2.x)
  • A tool call fails with no log — unclear whether the Server crashed or the arguments were wrong
  • Token costs spike with no visibility into which Server's which tool caused it

Registry solves discovery. Routing solves dispatch. Observability solves diagnosis.

MCP Registry

A Registry is the enterprise directory of MCP Servers: their locations, versions, capabilities, and owners.

# mcp-registry.yaml
servers:
  - id: jira-tools
    name: Jira Tools
    description: "Search, create, and update Jira tickets"
    version: "2.1.0"
    domain: engineering
    owner: "@team-platform"
    status: active
    transport: stdio
    command: python
    args: ["/opt/mcp/jira/server.py"]
    capabilities:
      tools: [search_issues, create_issue, update_issue]
      resources: [jira://projects, jira://sprint/current]
    metrics:
      monthly_calls: 4521
      avg_latency_ms: 180
      error_rate: 0.2%
 
  - id: github-tools
    name: GitHub Tools
    version: "1.5.0"
    domain: engineering
    owner: "@team-platform"
    status: active
    transport: stdio
    command: npx
    args: ["-y", "@modelcontextprotocol/server-github"]
    capabilities:
      tools: [create_pull_request, search_repositories, get_file_contents]
 
  - id: jira-tools-legacy
    name: Jira Tools (Legacy)
    version: "1.2.0"
    domain: engineering
    status: deprecated
    deprecation:
      reason: "Superseded by jira-tools v2.x; search_jira renamed to search_issues"
      migration_guide: "Replace search_jira with search_issues; argument structure unchanged"
      removal_date: "2026-10-01"
    capabilities:
      tools: [search_jira, create_jira_ticket]

The Registry solves three things:

  1. Discovery: new Agents check the Registry to learn what's available — no relying on hallway conversations
  2. Deprecation signaling: deprecated status plus a migration_guide gives Agent code a concrete migration path
  3. Ownership: every Server has an owner — when something breaks, you know who to contact

Tool Routing Strategies

When an Agent needs to "search Jira tickets," how does it find the right Server? Four strategies:

Strategy 1: Static Configuration (simplest)

Declare all Servers directly in Agent settings:

{
  "mcpServers": {
    "jira": {
      "command": "python",
      "args": ["/opt/mcp/jira/server.py"]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"]
    }
  }
}

Simple and predictable. Adding a new Server requires updating every Agent config file manually.

Strategy 2: Domain-Based Loading

Load only the Servers relevant to the current task:

DOMAIN_SERVERS = {
    "engineering": ["jira-tools", "github-tools", "gitlab-tools"],
    "data": ["postgres-readonly", "bigquery-tools"],
    "communication": ["slack-tools", "email-tools"],
}
 
def load_servers_for_task(task_type: str) -> list[dict]:
    domain = classify_task_domain(task_type)
    server_ids = DOMAIN_SERVERS.get(domain, [])
    registry = load_registry()
    return [
        s for s in registry["servers"]
        if s["id"] in server_ids and s["status"] == "active"
    ]

Reduces unnecessary Server startup overhead. Each Agent loads only the tools it needs.

Strategy 3: Embedding Routing (semantic matching)

The same approach as Skill Series Article 06 — embed Server descriptions, embed the user request, find the nearest neighbors:

def route_to_server(user_input: str, registry: list[dict]) -> list[str]:
    query_embedding = embedder.embed(user_input)
    scored = []
    for server in registry:
        desc_embedding = get_cached_embedding(server["id"], server["description"])
        score = cosine_similarity(query_embedding, desc_embedding)
        scored.append((server["id"], score))
 
    scored.sort(key=lambda x: x[1], reverse=True)
    return [s[0] for s in scored[:3] if s[1] > 0.6]

Good for 20+ Servers where new ones get added frequently. The same limitation applies as in Skill routing: Servers in the same domain cluster in embedding space. Use negative examples in descriptions to distinguish them.

Strategy 4: Hierarchical Routing (recommended)

Coarse-filter by domain first, then run embedding matching within the domain:

def hierarchical_route(user_input: str, registry: list[dict]) -> list[str]:
    # Layer 1: LLM classifies domain quickly
    domain = llm_classify_domain(user_input)   # "engineering" / "data" / ...
 
    # Layer 2: embedding match within domain
    domain_servers = [s for s in registry if s.get("domain") == domain]
    return embedding_route(user_input, domain_servers)

Observability: Langfuse Integration

Without Trace, MCP tool calls are a black box. When something fails, you don't know why. When latency spikes, you don't know where. When token costs grow, you don't know which tool caused it.

Three-Layer Trace Structure

from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
 
langfuse = Langfuse()
 
@observe(name="mcp_tool_call")
async def traced_tool_call(
    server_id: str,
    tool_name: str,
    arguments: dict,
    call_fn
) -> dict:
    langfuse_context.update_current_observation(
        input={"tool": tool_name, "arguments": arguments},
        metadata={
            "server_id": server_id,
            "server_version": get_server_version(server_id),
        }
    )
 
    t0 = time.perf_counter()
    try:
        result = await call_fn(tool_name, arguments)
        latency_ms = (time.perf_counter() - t0) * 1000
 
        langfuse_context.update_current_observation(
            output=result,
            metadata={"latency_ms": round(latency_ms, 2), "success": True}
        )
        return result
 
    except Exception as exc:
        latency_ms = (time.perf_counter() - t0) * 1000
        langfuse_context.update_current_observation(
            output={"error": str(exc)},
            metadata={"latency_ms": round(latency_ms, 2), "success": False},
            level="ERROR"
        )
        raise

Session-Level Trace

@observe(name="agent_session")
async def run_agent_session(user_input: str, session_id: str):
    langfuse_context.update_current_trace(
        session_id=session_id,
        user_id="user:alice",
        metadata={"task_type": "jira_query"}
    )
    result = await agent.run(user_input)
    langfuse_context.update_current_observation(output={"result": result})
    return result

Every tool call inside this session automatically attaches to the session trace.

What Trace Answers

Which tool call is slowest?
  → sort Spans by latency_ms
 
Which Server has the highest error rate?
  → group by server_id, count success=False fraction
 
Where is token spend concentrated?
  → LLM Span usage field, grouped by tool_name
 
What caused a specific Agent session failure?
  → search by session_id in Langfuse, expand the trace tree

Alert Rules

Wire key metrics into your alerting system (Grafana / Prometheus):

# Prometheus alerting rules
groups:
  - name: mcp_server_alerts
    rules:
      - alert: MCPServerHighErrorRate
        expr: mcp_tool_call_error_rate_5m > 0.05
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "MCP Server {{ $labels.server_id }} error rate {{ $value | humanizePercentage }}"
 
      - alert: MCPToolHighLatency
        expr: mcp_tool_call_p90_latency_ms > 5000
        for: 3m
        labels:
          severity: warning
        annotations:
          summary: "Tool {{ $labels.tool_name }} P90 latency {{ $value }}ms"
 
      - alert: DeprecatedToolInUse
        expr: mcp_deprecated_tool_calls_total > 0
        for: 0m
        labels:
          severity: info
        annotations:
          summary: "Deprecated tool {{ $labels.tool_name }} called {{ $value }} times today"

Three-Level Governance Roadmap

Level 1 — Individual/small team (do now):

  • Create mcp-registry.yaml with id / version / owner / status for every Server
  • Each Server's README lists its tools and usage examples

Level 2 — Team sharing (1-2 weeks):

  • Registry in Git, changes go through PR review
  • Langfuse Trace on every tool call: log tool_name / latency / success at minimum
  • Error rate alerts configured

Level 3 — Enterprise governance (ongoing):

  • Hierarchical routing (domain classification + embedding selection)
  • Monthly Server health report (call volume, error rate, top consumers)
  • Formal deprecation process (announce → alert → 90-day grace period → remove)

Summary

  1. Registry is the foundation of discovery: without it, 20 Servers rely on word-of-mouth, duplicate development and version confusion are inevitable
  2. Match routing strategy to scale: static config for 3 Servers; hierarchical routing (domain filter + embedding selection) for 20+, avoiding the same-domain embedding confusion problem
  3. Observability turns black boxes into records: Langfuse's session-level + tool-call-level Trace means failure root causes go from 'unknown' to 'found in 5 seconds on the Dashboard'

References

Check out PrimeSkills — a curated marketplace of AI agents and skills that have been validated in real-world, enterprise-grade workflows. No fluff, just what actually works.

Find more useful knowledge and interesting products on my Homepage

← Back to Blog