Appearance
5.2 Model Context Protocol (MCP) Architecture & Consumption
Canonical Curriculum Reference:
docs/plan/05_phase5_harness_engineering_mcp.md
Textbook Reference: AI Agents and Applications (Manning Publications), Chapter 13: "Building and consuming MCP servers" (pp. 336–354).
🌐 1. The
In traditional AI application engineering, connecting
The Model Context Protocol (MCP), open-sourced by Anthropic and rapidly adopted by Google, Microsoft, and the open-source community, solves this by establishing a standardized open protocol based on JSON-RPC 2.0:
Traditional: M x N Custom Glue Code MCP Standard: M + N Open Protocol
+----------+ +-----------+ +----------+ +-----------+
| Client 1 |<---->| Service 1 | | Client 1 |\ /| Service 1 |
+----------+ \ / +-----------+ +----------+ \ / +-----------+
\/ \ /
/\ [ MCP Bus ]
+----------+ / \ +-----------+ +----------+ /JSON-RPC\ +-----------+
| Client 2 |<---->| Service 2 | | Client 2 | / \ | Service 2 |
+----------+ +-----------+ +----------+ \+-----------+The Three Core MCP Primitives
MCP servers expose three distinct types of capabilities over transport layers (standard input/output stdio or Server-Sent Events SSE/HTTP):
- Tools (
tools/list,tools/call): Executable functions that allow models to take actions (e.g., query database, create PR, fetch live weather, run shell commands). - Resources (
resources/list,resources/read): File-like read-only data streams (e.g., source code files, database schemas, log records) that provide contextual grounding. - Prompts (
prompts/list,prompts/get): Pre-configured prompt templates and multi-turn workflows managed directly by the server.
🛠️ 2. Building Enterprise MCP Servers with FastMCP
FastMCP (Python) provides a high-level, declarative framework for building MCP servers with automatic Pydantic v2 schema derivation:
python
from typing import Annotated
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field, ConfigDict
# 1. Initialize FastMCP Server
mcp = FastMCP("EnterpriseInventoryServer")
class ProductQuery(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
sku: str = Field(pattern=r"^[A-Z]{3}-\d{4}$", description="SKU format: ABC-1234")
include_warehouse_breakdown: bool = Field(default=False)
# 2. Expose a strongly typed Tool
@mcp.tool()
def get_inventory_status(
query: Annotated[ProductQuery, "The validated product inventory query parameters"]
) -> str:
"""Queries enterprise SAP/ERP system for real-time stock levels."""
# Deterministic backend retrieval (e.g. Postgres or Redis)
return (
f"SKU {query.sku}: 1,450 units available across NA-East and EU-Central hubs."
)
# 3. Expose a Resource
@mcp.resource("inventory://schemas/current")
def get_inventory_schema() -> str:
"""Returns the JSON schema of the current inventory data warehouse."""
return ProductQuery.model_json_schema()
if __name__ == "__main__":
mcp.run(transport="stdio")🔌 3. Consuming Remote MCP Servers via MultiServerMCPClient
A production AI harness does not hardcode tool implementations. Instead, it dynamically subscribes to multiple remote MCP servers, discovers their schemas, and translates them directly into the agent's tool registry.
python
import asyncio
from typing import Any
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from pydantic import BaseModel, ConfigDict, Field
class MCPToolDefinition(BaseModel):
"""Normalized tool metadata derived from remote MCP server."""
model_config = ConfigDict(frozen=True)
name: str
description: str
parameters_schema: dict[str, Any]
server_name: str
class MultiServerMCPClient:
"""Orchestrates connections to multiple remote MCP servers over stdio or SSE."""
def __init__(self, server_configs: dict[str, StdioServerParameters]) -> None:
self._configs = server_configs
self._tools: dict[str, MCPToolDefinition] = {}
self._sessions: dict[str, ClientSession] = {}
async def connect_all(self) -> None:
"""Connects to all configured servers and retrieves dynamic toolsets."""
for server_name, params in self._configs.items():
# In production, manage persistent connection lifecycle & retries
read_stream, write_stream = await stdio_client(params).__aenter__()
session = await ClientSession(read_stream, write_stream).__aenter__()
await session.initialize()
self._sessions[server_name] = session
# Dynamically list tools
tools_result = await session.list_tools()
for tool in tools_result.tools:
namespaced_name = f"{server_name}_{tool.name}"
self._tools[namespaced_name] = MCPToolDefinition(
name=namespaced_name,
description=tool.description or "",
parameters_schema=tool.inputSchema or {},
server_name=server_name
)
async def call_tool(self, namespaced_name: str, arguments: dict[str, Any]) -> str:
"""Dispatches tool execution to the appropriate remote MCP server."""
if namespaced_name not in self._tools:
raise KeyError(f"Tool {namespaced_name} is not registered in active MCP client.")
tool_def = self._tools[namespaced_name]
session = self._sessions[tool_def.server_name]
original_name = namespaced_name.removeprefix(f"{tool_def.server_name}_")
result = await session.call_tool(name=original_name, arguments=arguments)
# Aggregate text contents
content_lines = [
content.text for content in result.content if hasattr(content, "text")
]
return "\n".join(content_lines)
@property
def registered_tools(self) -> list[MCPToolDefinition]:
return list(self._tools.values())🔗 4. Bridging MCP Tools into LangChain & LCEL
Once MCP tools are discovered, they can be adapted directly into standard LangChain StructuredTool objects, making them first-class components in LCEL chains and LangGraph agents:
python
from langchain_core.tools import StructuredTool
from pydantic import create_model
def adapt_mcp_tool_to_langchain(mcp_tool: MCPToolDefinition, client: MultiServerMCPClient) -> StructuredTool:
"""Dynamically converts an MCP Tool Definition into a LangChain StructuredTool."""
async def async_executor(**kwargs: Any) -> str:
return await client.call_tool(mcp_tool.name, kwargs)
def sync_executor(**kwargs: Any) -> str:
return asyncio.run(async_executor(**kwargs))
return StructuredTool(
name=mcp_tool.name,
description=mcp_tool.description,
func=sync_executor,
coroutine=async_executor,
args_schema=None # Can be dynamically synthesized from parameters_schema
)🔒 5. Zero-Trust Security for MCP Consumption
When subscribing to external or third-party MCP servers, the harness must apply Zero-Trust Security Principles:
- Tool Namespacing: Always prefix tool names with the server origin (
github_create_issue,aws_terminate_instance) to prevent tool name collision and malicious shadowing. - Schema Sanitization: Never execute unvalidated tool arguments; validate types against strict JSON schemas before transmitting payloads over the wire.
- Execution Sandbox: Execute local MCP stdio servers in unprivileged subprocesses with isolated directory access (
bubblewrapon Linux or sandbox containers).