Primitive Shifts: MCP Gateways
Every few months, the baseline of how AI systems work quietly moves. Engineers who noticed early weren't smarter — they were just paying attention to the right signals. The shift from monolithic prompts to tool-calling agents caught teams flat-footed in 2024. The move from single-agent demos to multi-agent production systems surprised people again in 2025. Right now, another floor is moving: the emergence of a governance layer between agents and the tools they invoke.
What Is It?
MCP Gateways are an infrastructure layer that sits between AI agents and MCP servers, providing centralized authentication, routing, rate limiting, and observability across all agent-to-tool connections. If you're running agents in production today, you're probably connecting them directly to MCP servers — your agent spawns a connection, calls a tool, gets results. That works fine until it doesn't.
The pattern mirrors what happened with REST microservices a decade ago. Individual service-to-service calls were technically functional but operationally ungovernable. Teams couldn't answer basic questions: Which service called which endpoint? How often? With what payloads? API gateways (Kong, Apigee, Ambassador) emerged to create a single chokepoint that made those calls auditable, throttled, and permission-aware.
MCP Gateways do the same thing for agent-to-tool traffic. They intercept every tool invocation before it reaches the destination server, enforce policies, log requests and responses, and propagate identity context downstream. The capabilities converging across implementations include OAuth/SSO propagation to downstream tools, request/response logging with full payload capture, cost attribution per agent/user/team, and circuit breakers for misbehaving tool servers.
What makes this a primitive shift rather than incremental tooling is the abstraction it introduces: separating "what tools exist" (MCP server registration) from "who can use them under what conditions" (gateway policy). The Context Kubernetes thesis argues that orchestration layers outlast the primitives they govern — the gateway is the orchestration layer for MCP, and it's becoming mandatory once organizations have more than a handful of agents in production.
Why It's Flying Under the Radar
MCP adoption dominated 2025 headlines. Every major model provider shipped MCP support. Anthropic's 2026 Agentic Coding Trends Report shows 78% of production agentic systems now use some form of tool calling. But the governance layer atop MCP is treated as "enterprise plumbing" rather than a paradigm shift, so it doesn't get the same attention.
Most teams with one to three agents don't feel the pain yet. Direct MCP connections work fine when you can manually track which agent has access to which tools, when you can eyeball logs during development, when the blast radius of a misbehaving agent is contained. The problem is that agent counts don't stay at three.
The major cloud providers haven't shipped first-party gateway products yet. Current implementations are either internal enterprise custom builds or early-stage open source projects, making them invisible to engineers scanning official docs. AWS, Azure, and GCP all have agent-related announcements, but none have branded "MCP Gateway" products with the marketing weight that drives adoption.
There's also genuine confusion with existing API gateways. Teams assume their current Kong or Envoy setup handles this, not realizing MCP's bidirectional streaming and tool-specific semantics require purpose-built middleware. A REST gateway doesn't understand that an MCP tool invocation carries an agent identity that needs to be validated against a policy before the tool receives the request.
The Position: Collaborative Agentic AI Needs Interoperability Across Ecosystems paper notes that protocol fragmentation (MCP vs. A2A vs. ACP) absorbs attention that should go to the shared need all these protocols have: a governed access layer that works across heterogeneous agent ecosystems.
Hands-On: Try It Today
The fastest way to understand what gateways provide is to build a minimal one. The following Python implementation creates a proxy that sits between any MCP client and server, logging every tool invocation and enforcing a simple policy: agents must present a valid token with specific claims to access tools tagged as "sensitive."
# mcp_gateway_proxy.py
# A minimal MCP Gateway demonstrating authentication, logging, and policy enforcement
# Requires: pip install mcp httpx pydantic python-jose
import asyncio
import json
import logging
from datetime import datetime
from typing import Optional
from dataclasses import dataclass, field
from jose import jwt, JWTError
# Configure structured logging for tool invocation audit trail
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s | %(levelname)s | %(message)s'
)
logger = logging.getLogger("mcp_gateway")
# Gateway configuration - in production, load from secrets manager
GATEWAY_CONFIG = {
"jwt_secret": "your-secret-key-replace-in-production",
"jwt_algorithm": "HS256",
"rate_limit_per_minute": 100,
"sensitive_tools": ["database_write", "file_delete", "admin_execute"],
"required_claims_for_sensitive": ["admin_access", "elevated_permissions"],
}
@dataclass
class InvocationLog:
"""Structured log entry for every tool invocation passing through the gateway."""
timestamp: str
agent_id: str
tool_name: str
request_payload: dict
response_payload: Optional[dict] = None
latency_ms: Optional[float] = None
policy_decision: str = "pending"
error: Optional[str] = None
@dataclass
class RateLimitState:
"""Per-agent rate limiting state - tracks invocations within sliding window."""
invocation_count: int = 0
window_start: datetime = field(default_factory=datetime.utcnow)
class MCPGateway:
"""
MCP Gateway proxy that intercepts tool invocations between agents and servers.
Provides: authentication, authorization, rate limiting, logging, circuit breaking.
"""
def __init__(self, upstream_mcp_url: str):
self.upstream_url = upstream_mcp_url
self.rate_limits: dict[str, RateLimitState] = {}
self.invocation_logs: list[InvocationLog] = []
self.circuit_breaker_open: dict[str, bool] = {}
def validate_agent_token(self, token: str) -> tuple[bool, dict]:
"""
Validate JWT token from agent, extract claims for policy decisions.
In production: integrate with your SSO provider (Okta, Auth0, etc.)
"""
try:
payload = jwt.decode(
token,
GATEWAY_CONFIG["jwt_secret"],
algorithms=[GATEWAY_CONFIG["jwt_algorithm"]]
)
return True, payload
except JWTError as e:
logger.warning(f"Token validation failed: {e}")
return False, {}
def check_policy(self, agent_claims: dict, tool_name: str) -> tuple[bool, str]:
"""
Evaluate access policy: can this agent invoke this tool?
This is where enterprise policies get enforced uniformly.
"""
# Check if tool requires elevated permissions
if tool_name in GATEWAY_CONFIG["sensitive_tools"]:
agent_permissions = agent_claims.get("permissions", [])
required = GATEWAY_CONFIG["required_claims_for_sensitive"]
if not any(perm in agent_permissions for perm in required):
return False, f"Tool '{tool_name}' requires one of {required}"
return True, "allowed"
def check_rate_limit(self, agent_id: str) -> tuple[bool, str]:
"""
Enforce per-agent rate limits - prevents runaway agents from overwhelming tools.
"""
now = datetime.utcnow()
if agent_id not in self.rate_limits:
self.rate_limits[agent_id] = RateLimitState()
state = self.rate_limits[agent_id]
# Reset window if minute has passed
elapsed = (now - state.window_start).total_seconds()
if elapsed > 60:
state.invocation_count = 0
state.window_start = now
if state.invocation_count >= GATEWAY_CONFIG["rate_limit_per_minute"]:
return False, f"Rate limit exceeded: {state.invocation_count}/min"
state.invocation_count += 1
return True, "within_limit"
async def proxy_invocation(
self,
agent_token: str,
tool_name: str,
tool_args: dict
) -> dict:
"""
Main gateway logic: authenticate, authorize, rate limit, proxy, log.
This single method captures the entire value proposition of the gateway.
"""
start_time = datetime.utcnow()
# Step 1: Validate agent identity
token_valid, claims = self.validate_agent_token(agent_token)
if not token_valid:
return {"error": "authentication_failed", "details": "Invalid agent token"}
agent_id = claims.get("agent_id", "unknown")
# Initialize log entry - we capture everything, success or failure
log_entry = InvocationLog(
timestamp=start_time.isoformat(),
agent_id=agent_id,
tool_name=tool_name,
request_payload=tool_args,
)
# Step 2: Check rate limits
rate_ok, rate_msg = self.check_rate_limit(agent_id)
if not rate_ok:
log_entry.policy_decision = "rate_limited"
log_entry.error = rate_msg
self.invocation_logs.append(log_entry)
logger.warning(f"Rate limited: {agent_id} on {tool_name}")
return {"error": "rate_limited", "details": rate_msg}
# Step 3: Evaluate access policy
policy_ok, policy_msg = self.check_policy(claims, tool_name)
if not policy_ok:
log_entry.policy_decision = "denied"
log_entry.error = policy_msg
self.invocation_logs.append(log_entry)
logger.warning(f"Policy denied: {agent_id} -> {tool_name}: {policy_msg}")
return {"error": "policy_denied", "details": policy_msg}
# Step 4: Check circuit breaker for upstream tool server
if self.circuit_breaker_open.get(tool_name, False):
log_entry.policy_decision = "circuit_open"
log_entry.error = "Upstream tool server unavailable"
self.invocation_logs.append(log_entry)
return {"error": "circuit_open", "details": "Tool temporarily unavailable"}
# Step 5: Proxy to upstream MCP server (simulated here)
# In production: use mcp client library to forward the invocation
try:
# Simulate upstream call - replace with actual MCP client forwarding
response = await self._forward_to_upstream(tool_name, tool_args, claims)
end_time = datetime.utcnow()
latency = (end_time - start_time).total_seconds() * 1000
log_entry.response_payload = response
log_entry.latency_ms = latency
log_entry.policy_decision = "allowed"
logger.info(
f"Invocation: {agent_id} -> {tool_name} | "
f"{latency:.1f}ms | allowed"
)
except Exception as e:
log_entry.policy_decision = "upstream_error"
log_entry.error = str(e)
self.invocation_logs.append(log_entry)
# Trigger circuit breaker after repeated failures (simplified)
logger.error(f"Upstream error for {tool_name}: {e}")
return {"error": "upstream_error", "details": str(e)}
self.invocation_logs.append(log_entry)
return response
async def _forward_to_upstream(
self,
tool_name: str,
tool_args: dict,
agent_claims: dict
) -> dict:
"""
Forward invocation to upstream MCP server with propagated identity.
The gateway adds agent context so downstream tools can make their own
authorization decisions without trusting agent self-reported claims.
"""
# In production, this would use the mcp library to connect to upstream
# Here we simulate the response
await asyncio.sleep(0.05) # Simulate network latency
return {
"result": f"Executed {tool_name}",
"args_received": tool_args,
"executed_for": agent_claims.get("agent_id"),
"gateway_verified": True # Downstream knows gateway validated identity
}
def get_audit_report(self) -> dict:
"""
Generate audit report from captured logs - this visibility is the gateway's
core value for security and compliance teams.
"""
tool_counts = {}
agent_counts = {}
denied_count = 0
for log in self.invocation_logs:
tool_counts[log.tool_name] = tool_counts.get(log.tool_name, 0) + 1
agent_counts[log.agent_id] = agent_counts.get(log.agent_id, 0) + 1
if log.policy_decision in ["denied", "rate_limited"]:
denied_count += 1
return {
"total_invocations": len(self.invocation_logs),
"denied_invocations": denied_count,
"invocations_by_tool": tool_counts,
"invocations_by_agent": agent_counts,
"logs": [vars(log) for log in self.invocation_logs[-10:]] # Last 10
}
# Example usage demonstrating the gateway in action
async def main():
gateway = MCPGateway(upstream_mcp_url="http://localhost:8080/mcp")
# Create test tokens with different permission levels
regular_token = jwt.encode(
{"agent_id": "agent-001", "permissions": ["read", "write"]},
GATEWAY_CONFIG["jwt_secret"],
algorithm=GATEWAY_CONFIG["jwt_algorithm"]
)
admin_token = jwt.encode(
{"agent_id": "agent-admin", "permissions": ["admin_access"]},
GATEWAY_CONFIG["jwt_secret"],
algorithm=GATEWAY_CONFIG["jwt_algorithm"]
)
# Test 1: Regular agent accessing normal tool - should succeed
result = await gateway.proxy_invocation(
regular_token, "file_read", {"path": "/data/report.txt"}
)
print(f"Regular tool access: {result}")
# Test 2: Regular agent accessing sensitive tool - should be denied
result = await gateway.proxy_invocation(
regular_token, "database_write", {"table": "users", "data": {}}
)
print(f"Sensitive tool (regular agent): {result}")
# Test 3: Admin agent accessing sensitive tool - should succeed
result = await gateway.proxy_invocation(
admin_token, "database_write", {"table": "users", "data": {}}
)
print(f"Sensitive tool (admin agent): {result}")
# Print audit report
print("\n--- Audit Report ---")
print(json.dumps(gateway.get_audit_report(), indent=2))
if __name__ == "__main__":
asyncio.run(main())
Run this code, then modify the policy rules. Add tools to the sensitive list. Adjust rate limits. Watch how the audit report captures everything. The exercise reveals how naked most current MCP deployments are — they have no equivalent of this policy layer, no audit trail, no rate limiting that happens before the tool even receives the request.
What This Means for Your Stack
Direct agent-to-MCP-server connections become technical debt the moment a second team wants to use the same tool with different permissions. Retrofitting a gateway later is harder than starting with one because you've already baked assumptions about direct connectivity into your agent code.
Security teams will require gateway-level audit logs before approving production agent deployments. The 2026 Agentic Coding Trends Report shows security review cycles for AI agents averaging 4.2 weeks, with "audit trail completeness" as the most common blocker. Teams without a gateway layer will face compliance blockers they didn't anticipate.
Cost management shifts from model-level token counting to tool-level invocation accounting. You already track how many tokens your agents consume. But do you know how many database queries they trigger? How many API calls to third-party services? The gateway is the only place with visibility into both sides of that equation.
Multi-agent systems that share tools need coordination primitives — locks, quotas, priority queues — that belong in the gateway, not replicated across every agent's codebase. How agents are transforming work describes scenarios where dozens of agents share access to the same CRM or ticketing system. Without central coordination, you get race conditions, quota exhaustion, and unpredictable failures.
The gateway becomes the natural integration point for agent identity. The best AI agent frameworks in 2026 notes that identity propagation — flowing user context through agent chains to downstream tools — is "the most underspecified part of current architectures." The gateway solves this by validating agent tokens once and propagating verified claims downstream.
The Infrastructure Signal
The infrastructure signals are converging faster than most teams realize. AWS's Agent Registry (April 2026 preview) provides centralized discovery and approval workflows for agents, tools, and MCP servers — a control plane that implies governed access, not raw connections. You don't build a registry unless you expect to control what's in it.
Microsoft's Agent Framework at BUILD 2026 introduced ToolApprovalAgent as a first-class component. This isn't a library — it's a runtime primitive that expects to intercept tool calls and evaluate policies before execution. The major players are building gateway patterns into their agent infrastructure.
The Linux Foundation's acceptance of MCP stewardship — with Anthropic, OpenAI, Microsoft, and Google participating — creates pressure for interoperability standards. AI Governance and Regulation 2026: A Complete Guide tracks how regulatory requirements are pushing toward standardized audit trails for AI system actions. Gateways will need to implement these standards; teams building them now will have a head start.
Enterprise incident patterns are emerging that trace directly to ungoverned tool access. Supply chain vulnerabilities and data exfiltration via prompt injection both exploit the same gap: agents with unfettered access to tools that trust the agent's self-reported context. A gateway layer that validates identity and enforces policies would have intercepted these attacks.
An Empirical Study of Testing Practices in Open Source AI Agent Frameworks found that tool invocation testing is the weakest area of current agent test suites. The gateway becomes the natural place to inject test doubles, simulate failures, and capture interactions for replay — capabilities that dramatically improve testability of agent systems.
Shift Rating
🟡 Experiment
MCP Gateways are not yet standardized enough for blind adoption. The specification isn't finalized. First-party cloud products don't exist. The Position: Collaborative Agentic AI Needs Interoperability Across Ecosystems paper explicitly calls out the lack of standardized agent identity and tool access governance as a blocking issue for enterprise adoption.
But teams running more than two agents in production should prototype a gateway layer now. Deploy the code above in front of one MCP server. Capture a week of logs. Build a dashboard. You'll discover tool invocation patterns you've never seen before — agents calling tools you didn't expect, at frequencies that surprise you, with payloads that reveal assumptions baked into your prompts.
Within 12 months, ungoverned MCP connections will be viewed the same way unauthed REST endpoints were in 2015 — technically functional, professionally unacceptable. The teams experimenting today will have policy patterns, observability pipelines, and governance muscle memory that late adopters will spend months rebuilding under pressure. The floor is moving. The question is whether you notice before or after it's already moved.
Sources
- Context Kubernetes: Declarative Orchestration of Enterprise Knowledge for Agentic AI Systems
- Position: Collaborative Agentic AI Needs Interoperability Across Ecosystems
- An Empirical Study of Testing Practices in Open Source AI Agent Frameworks and Agentic Applications
- 2026 Agentic Coding Trends Report
- AI Governance and Regulation 2026: A Complete Guide
- How agents are transforming work
- The best AI agent frameworks in 2026
This is part of **Primitive Shifts* — a monthly series tracking when new AI building blocks
move from novel experiments to infrastructure you'll be expected to know.*
Follow the Next MCP Watch series on Dev.to to catch every edition.
Spotted a shift happening in your stack? Drop it in the comments.
Top comments (0)