Phase 5: Model Context Protocol (MCP) · 75 min · Python · MCP SDK · Anthropic SDK
The Problem
Your company has 50 internal MCP servers. A customer support agent needs database and email tools. An engineering agent needs GitHub and Postgres tools. A data analyst agent needs analytics and reporting tools.
Connecting every agent to all 50 servers at startup means loading 250+ tool definitions into every model call. Tool definitions consume tokens. At 250 tools, model performance degrades significantly — too many choices impairs decision-making. The model spends reasoning budget evaluating irrelevant tools instead of solving the task.
The solution is a tool registry: a catalog of available MCP servers tagged with capability labels. Each agent queries the registry for servers relevant to its current task, connects only to those, and loads a focused set of 5–10 tools. New servers join the registry without changing any agent code. Teams see only the tools they're authorized to access.
Demand-driven tool loading
Instead of connecting to all servers upfront, the agent asks: "What capabilities does this task need?" — then queries the registry for matching servers. This keeps tool sets small, model decisions focused, and context windows efficient.
Registry architecture
An MCPRegistry stores server metadata as catalog entries:
@dataclass
class ServerRegistration:
server_id: str
name: str
description: str
capabilities: list[str] # e.g. ["database", "sql", "read-only"]
endpoint: str # stdio command or HTTP URL
transport: str # "stdio" | "streamable-http"
auth_required: bool
allowed_teams: list[str] # ["*"] = all teams, else ["eng", "support"]
status: str # "available" | "unavailable"
The registry answers two queries:
query_by_capability(["database", "sql"], team="eng") → returns servers that have any of these capabilities and are accessible to the eng teamlist_all(team="support") → all servers a team can access
Why load tools dynamically from a registry instead of connecting to all servers upfront?
Tool definitions consume tokens. 50 servers × 5 tools each = 250 tool descriptions in every model call. Beyond ~20–30 tools, model performance degrades — too many choices impairs decision-making. Dynamic discovery loads only the 5–10 tools actually needed for the current task, keeping the context focused and the model performant.
Capability-based selection
Each server is tagged with capability labels: ["database", "sql", "read-only"], ["filesystem", "read", "write"], ["web", "search", "browser"]. Capability labels describe what type of work the server enables, not its specific tools.
The CapabilityMatcher extracts required capabilities from a task description. You can implement this two ways:
- LLM extraction: ask Claude to classify the task — "Given this task, which capability labels from this list are needed?" Fast, flexible, handles novel phrasing.
- Keyword matching: regex or keyword lookup — map "query", "database", "SQL" →
["database", "sql"]. Simple, no API call needed, slightly brittle.
Then query the registry: registry.query_by_capability(["database", "sql"], team="eng") returns the two database servers accessible to engineering. The agent connects to those two, loads their 10 tools, and runs.
Multi-tenant access control
In enterprise deployments, different teams have access to different MCP servers. An HR agent shouldn't be able to call the engineering team's deployment tools. An engineering agent shouldn't automatically have access to HR's payroll server.
The registry stores per-server ACLs:
ServerRegistration(
server_id="deployment-api",
name="Deployment API",
description="Manages deployments and infrastructure",
capabilities=["deployment", "infrastructure"],
endpoint="https://deploy-api.internal:8080",
transport="streamable-http",
auth_required=True,
allowed_teams=["eng", "devops"], # NOT "support" or "hr"
status="available",
)
When an agent connects as team="support", query_by_capability(caps, team="support") filters out any server where "support" is not in allowed_teams (and "*" is not present either). The support agent never even sees the deployment tools — they're invisible to its tool discovery.
A multi-tenant MCP registry with ACLs (access control lists) solves what problem?
In enterprise settings, access control is critical. An MCP registry with team-level ACLs ensures that agents operating in different contexts only receive tool definitions for tools they're authorized to use. This is the same principle as role-based access control (RBAC) in traditional systems — applied to AI tool access.
Registry health checks
Before an agent run, ping all registered servers to mark unavailable ones. An agent that tries to connect to an unavailable server wastes time on timeouts.
async def health_check(self) -> dict[str, str]:
for server_id, server in self._servers.items():
try:
# In production: connect and call initialize, expect a response
conn = await connect_server(server.endpoint, timeout=2.0)
await conn.initialize()
server.status = "available"
except Exception:
server.status = "unavailable"
return {sid: s.status for sid, s in self._servers.items()}
Run health checks at agent startup and periodically (every few minutes in long-running agents). The registry's query_by_capability() automatically excludes unavailable servers from results.
Build
Build it
This standalone simulation captures the core registry pattern: capability tags, team ACLs, and a demand-driven lookup — no MCP transport needed to see the idea work.
registry = [
{"id": "db-eng", "capabilities": ["database", "sql"], "teams": ["eng"]},
{"id": "db-support", "capabilities": ["database", "sql"], "teams": ["support"]},
{"id": "email", "capabilities": ["email", "notify"], "teams": ["*"]},
{"id": "deploy-api", "capabilities": ["deployment", "infrastructure"], "teams": ["eng", "devops"]},
]
def query_by_capability(caps, team):
matches = []
for server in registry:
has_capability = any(c in server["capabilities"] for c in caps)
has_access = "*" in server["teams"] or team in server["teams"]
if has_capability and has_access:
matches.append(server["id"])
return matches
tasks = [
(["database", "sql"], "eng"),
(["database", "sql"], "support"),
(["deployment"], "support"),
(["email"], "support"),
]
for caps, team in tasks:
servers = query_by_capability(caps, team)
print(f"team={team:8} needs={caps} -> visible servers: {servers}")
The code artifact lesson_05_04_registry.py implements:
MCPRegistry with capability-based lookup and team ACLsCapabilityMatcher with LLM extraction and keyword fallbackDynamicMCPAgent that queries the registry per task and connects only to needed servers- Multi-tenant demo: 4 teams with different server visibility
- Health check simulation: marks unavailable servers before agent runs
Run it:
pip install mcp anthropic
python lesson_05_04_registry.py