Building Custom MCP Clients
A comprehensive, production-grade guide to designing, implementing, and scaling custom Model Context Protocol (MCP) clients using Python, langchain-mcp-adapters, and modern agentic LLM runtimes.
1. The Need for Custom MCP Clients
The Architectural Limitation of Default Hosts
In standard MCP setups, developers routinely rely on pre-packaged host environments like Claude Desktop. While turnkey hosts are valuable for local experimentation, they confine MCP capabilities inside third-party graphical applications.
Building enterprise conversational agents, proprietary internal tools, or domain-specific chat surfaces requires direct, programmable control over the MCP client runtime.
Structure & Concept Flow
Core Motivations
- Decoupled Architecture: Separate the agent reasoning engine, client aggregation layer, and user interface.
- Multi-Server Aggregation: Dynamically merge local system tools and cloud-hosted remote tools under a unified model interface.
- Custom Application Logic: Control message state, execution inspection, multi-tool loops, and UI presentation without proprietary desktop vendor boundaries.
2. Theoretical Architecture & Mental Models
The MCP Triad: Host, Client, and Server
The Model Context Protocol establishes strict role boundaries:
- MCP Host: The application environment running the agent loops, user interface, and overall system lifecycle (e.g., your Python runtime or Streamlit application).
- MCP Client: The protocol engine inside the host responsible for initiating connections, parsing manifests, negotiating protocol versions, and executing tool calls.
- MCP Server: The independent process exposing tools, resources, and prompt templates over standard transports.
Client-Side Invocation & Resolution Flow
The sequence below illustrates how an MCP client acts as the central router between the LLM and disparate MCP servers:
Structure & Concept Flow
3. Tooling Ecosystem & Library Selection
When implementing an MCP client in Python, several libraries exist across the ecosystem:
Why langchain-mcp-adapters?
- Universal LLM Interoperability: Converts MCP tool definitions into OpenAI/Anthropic tool schemas out of the box using
client.get_tools(). - LangGraph Ready: Directly embeds into cyclical graph runtimes without manual adapter logic.
- Multi-Server Support: Includes
MultiServerMCPClient, which simplifies connecting to both local child processes (stdio) and remote web services (sseorstreamable-http) in a single configuration block.
4. Environment Setup & Project Scaffolding
Modern Python environments benefit from using uv for deterministic dependency resolution and rapid subshell management.
Project Initialization
BASH
# Install uv globally if not already present
pip install uv
# Create and enter the project directory
mkdir mcp-client-workspace
cd mcp-client-workspace
# Initialize virtual environment and project manifest
uv init .
# Add core runtime dependencies
uv add langchain langchain-openai langchain-mcp-adapters python-dotenv streamlit
Environment Configuration
Create a .env file in the project root containing your model provider credentials:
BASH
OPENAI_API_KEY=your_actual_openai_api_key_here
5. Single-Server Client Implementation (stdio)
Connecting to a Local Child Process
When working with local servers, the MCP client spawns the server as a child process via stdio. The client configuration dictionary specifies the system command and argument vector used to start the process.
Create client_step1.py:
PYTHON
import asyncio
import os
from dotenv import load_dotenv
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, ToolMessage
load_dotenv()
# Define server parameters
# stdio transport requires the executable and arguments to spawn the process
SERVER_CONFIG = {
"math": {
"transport": "stdio",
"command": "uv",
"args": [
"run",
"--with",
"fastmcp",
"fastmcp",
"run",
"/absolute/path/to/math_server/main.py",
],
}
}
async def main():
# 1. Initialize client with multi-server registry
async with MultiServerMCPClient(SERVER_CONFIG) as client:
# 2. Fetch all registered tools from the running server process
tools = await client.get_tools()
named_tools = {tool.name: tool for tool in tools}
# 3. Bind discovered tools to the LLM
model = ChatOpenAI(model="gpt-4o", temperature=0)
model_with_tools = model.bind_tools(tools)
prompt = "Using your math tools, compute the product of 12 and 15."
messages = [HumanMessage(content=prompt)]
# 4. First inference step: Evaluate if tool calls are requested
response = await model_with_tools.ainvoke(messages)
messages.append(response)
# 5. Conditional branching: Direct answer vs. tool routing
if not response.tool_calls:
print("Direct Response:", response.content)
return
# 6. Execute requested tools
for tool_call in response.tool_calls:
tool_name = tool_call["name"]
tool_args = tool_call["args"]
tool_id = tool_call["id"]
selected_tool = named_tools[tool_name]
tool_output = await selected_tool.ainvoke(tool_args)
# Append the standard ToolMessage to maintain conversation state
messages.append(
ToolMessage(
content=str(tool_output),
tool_call_id=tool_id,
)
)
# 7. Final synthesis step: Provide full context back to the LLM
final_response = await model_with_tools.ainvoke(messages)
print("Final Output:", final_response.content)
if __name__ == "__main__":
asyncio.run(main())
6. Scaling to Multi-Server Architectures
Production assistants typically require more than one integration. MCP handles this cleanly: local system commands can execute side-by-side with remote cloud endpoints.
Structure & Concept Flow
Multi-Transport Registration Pattern
Update the configuration dictionary to register multiple servers simultaneously using different transport backends:
PYTHON
MULTI_SERVER_CONFIG = {
# Local Server: Spawned via command line over standard I/O
"math": {
"transport": "stdio",
"command": "uv",
"args": [
"run",
"--with",
"fastmcp",
"fastmcp",
"run",
"/Users/username/Desktop/mcp-servers/math_server/main.py"
]
},
# Remote Server: Communicates over network endpoints via Server-Sent Events / HTTP
"expense": {
"transport": "streamable-http",
"url": "https://expense-tracker-prod.mcp.cloud/mcp"
},
# Third-Party Local Engine: Community Manim Animation Server
"manim": {
"transport": "stdio",
"command": "uv",
"args": [
"run",
"manim-mcp-server"
]
}
}
When await client.get_tools() executes, MultiServerMCPClient performs parallel handshakes across all defined endpoints, aggregating their tools into a single flat list:
PYTHON
async with MultiServerMCPClient(MULTI_SERVER_CONFIG) as client:
all_tools = await client.get_tools()
# all_tools now contains:
# ['add', 'subtract', 'multiply', 'divide', 'modulo',
# 'add_expense', 'list_expenses', 'summarize_expenses',
# 'render_animation', 'clean_directory']
7. Enterprise UI Layer: Streamlit Conversational Host
To transition from terminal scripts to a production interface, wrap the client in an interactive frontend. The example below uses Streamlit to manage chat state, handle async execution, and render dynamic output (such as generated video files).
Save this as app.py:
PYTHON
import asyncio
import os
import streamlit as st
from dotenv import load_dotenv
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
load_dotenv()
st.set_page_config(page_title="Universal MCP Assistant", layout="wide")
st.title("Enterprise Multi-Server MCP Agent")
# Define all target servers
SERVER_CONFIGS = {
"math": {
"transport": "stdio",
"command": "uv",
"args": ["run", "--with", "fastmcp", "fastmcp", "run", "./servers/math_server.py"],
},
"expense": {
"transport": "streamable-http",
"url": "https://mcp-expense-instance.cloud/mcp",
}
}
# Maintain conversation history across Streamlit reruns
if "chat_history" not in st.session_state:
st.session_state.chat_history = []
# Display prior dialogue
for msg in st.session_state.chat_history:
if isinstance(msg, HumanMessage):
with st.chat_message("user"):
st.markdown(msg.content)
elif isinstance(msg, AIMessage) and msg.content:
with st.chat_message("assistant"):
st.markdown(msg.content)
async def process_user_query(query: str):
async with MultiServerMCPClient(SERVER_CONFIGS) as client:
tools = await client.get_tools()
named_tools = {tool.name: tool for tool in tools}
model = ChatOpenAI(model="gpt-4o", temperature=0)
model_with_tools = model.bind_tools(tools)
# Build context from session history + current prompt
invocation_chain = list(st.session_state.chat_history)
user_message = HumanMessage(content=query)
invocation_chain.append(user_message)
# Initial model inference
response = await model_with_tools.ainvoke(invocation_chain)
invocation_chain.append(response)
# Check if the model requested any tool operations
if response.tool_calls:
for call in response.tool_calls:
tool_name = call["name"]
tool_args = call["args"]
tool_id = call["id"]
with st.status(f"Executing `{tool_name}`...", expanded=False):
st.write("Arguments:", tool_args)
target_tool = named_tools[tool_name]
tool_result = await target_tool.ainvoke(tool_args)
st.write("Result:", tool_result)
invocation_chain.append(
ToolMessage(content=str(tool_result), tool_call_id=tool_id)
)
# Final response synthesis with tool outputs included
final_response = await model_with_tools.ainvoke(invocation_chain)
return user_message, final_response
else:
return user_message, response
# Chat input processing
user_input = st.chat_input("Enter your request...")
if user_input:
with st.chat_message("user"):
st.markdown(user_input)
with st.chat_message("assistant"):
with st.spinner("Processing through MCP Client..."):
user_msg, ai_msg = asyncio.run(process_user_query(user_input))
st.markdown(ai_msg.content)
# Check if a video was generated by an animation tool
if os.path.exists("media/output.mp4"):
st.video("media/output.mp4")
# Persist the interaction
st.session_state.chat_history.append(user_msg)
st.session_state.chat_history.append(ai_msg)
8. Common Pitfalls & Debugging Guide
1. The Missing tool_call_id Error
- Symptom: OpenAI or Anthropic APIs reject follow-up requests with errors like
InvalidParameter: tool_call_id missing or invalid. - Cause: Failing to attach the unique
tool_call_idgenerated during the model's initial invocation step when returning aToolMessage. - Resolution: Always extract
call["id"]directly from each item inresponse.tool_callsand assign it explicitly:ToolMessage(content=..., tool_call_id=call["id"]).
2. Subprocess Command Path Failures
- Symptom:
FileNotFoundError: [Errno 2] No such file or directory: 'uv'or unexpected subshell exits on launch. - Cause: When running servers locally via
stdio, desktop hosts and IDEs might use differing shellPATHvariables. - Resolution: Use fully qualified paths for binaries (e.g.,
/usr/local/bin/uvorC:\Users\...\uv.exe) and absolute paths for scripts in the configuration dictionary.
3. Asynchronous Loop Collisions
- Symptom:
RuntimeError: This event loop is already running. - Cause: Frameworks like Streamlit or Jupyter often run their own event loops. Calling
asyncio.run()inside an active loop causes collisions. - Resolution: Use
asyncio.get_event_loop().run_until_complete()or leveragenest_asynciowhen running inside environments with pre-existing loops:PYTHONimport nest_asyncio nest_asyncio.apply()
9. Summary and Key Takeaways
- Moving Beyond Turnkey Hosts: Building your own MCP client breaks the dependency on third-party applications (like Claude Desktop), giving you full control over models, tools, and interfaces.
- The LangChain Adapter Advantage: The
langchain-mcp-adapterspackage bridges MCP definitions with standard agent runtimes, converting remote protocol endpoints into native LangChain Structured Tools. - Unified Multi-Server Routing: A single
MultiServerMCPClientcan simultaneously orchestrate local processes running overstdioand remote cloud endpoints running overstreamable-http. - Strict Protocol State Management: Handling tool calls requires a consistent sequence: capture the model's requested call, execute the tool by name, package the result with its
tool_call_id, and send the complete history back to the model for final synthesis. - Handling Multi-Tool Responses: Production agents must parse
tool_callsiteratively within a loop rather than assuming only a single tool is invoked per turn. - UI-Agnostic Core Logic: By decoupling the agent's MCP execution logic from the presentation layer, you can reuse the same core backend across CLI utilities, interactive web apps, or backend REST APIs.