Content
# Model Context Protocol (MCP) Quick Start Guide
[TOC]
## Introduction
The Model Context Protocol (MCP) is an innovative open-source protocol that redefines the way large language models (LLM) interact with the external world. MCP provides a standardized approach that allows any large language model to easily connect to various data sources and tools, enabling seamless access to and processing of information. MCP acts like a USB-C interface for AI applications, providing a standardized way for AI models to connect to different data sources and tools.
MCP has several core features:
- Resources
- Prompts
- Tools
- Sampling
- Roots
- Transports
Since most features actually serve the Claude client, this document focuses on developing an MCP server that works with general large language models, emphasizing "tools" while briefly explaining other features at the end.
The transport layer of MCP supports the implementation of two protocols: stdio (standard input/output) and SSE (Server-Sent Events). Since stdio is more commonly used, this document will use stdio as an example for explanation.
This document will use Python version 3.11 and manage the Python project with uv. The code will be available on GitHub at the end of this document. Without further ado, let's get started!
## Developing an MCP Server
In this section, we will implement a server for web searching. First, let's initialize our project using uv.
> Official uv documentation: https://docs.astral.sh/uv/
```shell
# Initialize the project
uv init mcp_getting_started
cd mcp_getting_started
# Create a virtual environment and activate it
uv venv
.venv\Scripts\activate.bat
# Install dependencies
uv add "mcp[cli]" httpx openai
```
Next, we will create a file named `web_search.py` to implement our service. MCP provides us with two objects: `mcp.server.FastMCP` and `mcp.server.Server`. We will use the higher-level wrapper, `mcp.server.FastMCP`.
```python
import httpx
from mcp.server import FastMCP
# Initialize FastMCP server
app = FastMCP('web-search')
```
Implementing the execution method is very simple. MCP provides us with a `@mcp.tool()` decorator, and we just need to decorate our implementation function with it. The function name will serve as the tool name, the parameters will serve as tool parameters, and comments will describe the tool, parameters, and return values.
Here, we will directly use the Zhiyu interface, which not only helps us search for relevant result links but also generates summaries of the articles corresponding to those links. Currently, it is free, making it very suitable for our needs.
> Official documentation: https://bigmodel.cn/dev/api/search-tool/web-search-pro
>
> API Key generation address: https://bigmodel.cn/usercenter/proj-mgmt/apikeys
```python
@app.tool()
async def web_search(query: str) -> str:
"""
Search internet content
Args:
query: Content to search for
Returns:
Summary of search results
"""
async with httpx.AsyncClient() as client:
response = await client.post(
'https://open.bigmodel.cn/api/paas/v4/tools',
headers={'Authorization': 'Replace with your own API KEY'},
json={
'tool': 'web-search-pro',
'messages': [
{'role': 'user', 'content': query}
],
'stream': False
}
)
res_data = []
for choice in response.json()['choices']:
for message in choice['message']['tool_calls']:
search_results = message.get('search_result')
if not search_results:
continue
for result in search_results:
res_data.append(result['content'])
return '\n\n\n'.join(res_data)
```
Finally, let's add the code to run the server.
```python
if __name__ == "__main__":
app.run(transport='stdio')
```
## Debugging the MCP Server
At this point, we have completed the MCP server implementation. Next, we will use the official `Inspector` visualization tool to debug our server.
We can run `Inspector` in two ways:
> Please ensure that the node environment is installed.
Using npx:
```shell
npx -y @modelcontextprotocol/inspector <command> <arg1> <arg2>
```
The command to run our code is:
```shell
npx -y @modelcontextprotocol/inspector uv run web_search.py
```
Using mcp dev to run:
```shell
mcp dev PYTHONFILE
```
The command to run our code is:
```shell
mcp dev web_search.py
```
When the following prompt appears, it indicates that the run was successful.
Then, we open this address and click the `Connect` button on the left to connect to the service we just wrote. Next, we switch to the `Tools` tab and click the `List Tools` button to see the tool we just created, and we can start debugging!
## Developing an MCP Client
First, let's see how to call the tools from the MCP server we just developed on the client side.
```python
import asyncio
from mcp.client.stdio import stdio_client
from mcp import ClientSession, StdioServerParameters
# Create server parameters for stdio connection
server_params = StdioServerParameters(
# Command to execute the server, here we use uv to run web_search.py
command='uv',
# Running parameters
args=['run', 'web_search.py'],
# Environment variables, default is None, indicating to use current environment variables
# env=None
)
async def main():
# Create stdio client
async with stdio_client(server_params) as (stdio, write):
# Create ClientSession object
async with ClientSession(stdio, write) as session:
# Initialize ClientSession
await session.initialize()
# List available tools
response = await session.list_tools()
print(response)
# Call tool
response = await session.call_tool('web_search', {'query': 'What is the weather like in Hangzhou today?'})
print(response)
if __name__ == '__main__':
asyncio.run(main())
```
Since our Python script needs to run in a virtual environment, we will use `uv` to start our script.
Next, let's look at a small example of how to use `DeepSeek` to call methods from our MCP server.
We will use `dotenv` to manage our related environment variables. The content of the .env file is as follows:
```shell
OPENAI_API_KEY=sk-89baxxxxxxxxxxxxxxxxxx
OPENAI_BASE_URL=https://api.deepseek.com
OPENAI_MODEL=deepseek-chat
```
First, we will write our `MCPClient` class.
```python
import json
import asyncio
import os
from typing import Optional
from contextlib import AsyncExitStack
from openai import OpenAI
from dotenv import load_dotenv
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
load_dotenv()
class MCPClient:
def __init__(self):
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()
self.client = OpenAI()
```
Next, we add the `connect_to_server` method to initialize our MCP server session.
```python
async def connect_to_server(self):
server_params = StdioServerParameters(
command='uv',
args=['run', 'web_search.py'],
env=None
)
stdio_transport = await self.exit_stack.enter_async_context(
stdio_client(server_params))
stdio, write = stdio_transport
self.session = await self.exit_stack.enter_async_context(
ClientSession(stdio, write))
await self.session.initialize()
```
Then we implement a method to call the MCP server to handle interactions with DeepSeek.
```python
async def process_query(self, query: str) -> str:
# Here we need to constrain the large language model through a system prompt,
# otherwise it may not call the tool and provide random answers
system_prompt = (
"You are a helpful assistant."
"You have the function of online search. "
"Please MUST call web_search tool to search the Internet content before answering."
"Please do not lose the user's question information when searching,"
"and try to maintain the completeness of the question content as much as possible."
"When there is a date related question in the user's question,"
"please use the search function directly to search and PROHIBIT inserting specific time."
)
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": query}
]
# Get all MCP server tool list information
response = await self.session.list_tools()
# Generate function call description information
available_tools = [{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"input_schema": tool.inputSchema
}
} for tool in response.tools]
# Request DeepSeek, function call description information is passed through tools parameter
response = self.client.chat.completions.create(
model=os.getenv("OPENAI_MODEL"),
messages=messages,
tools=available_tools
)
# Process the returned content
content = response.choices[0]
if content.finish_reason == "tool_calls":
# If a tool is needed, parse the tool
tool_call = content.message.tool_calls[0]
tool_name = tool_call.function.name
tool_args = json.loads(tool_call.function.arguments)
# Execute the tool
result = await self.session.call_tool(tool_name, tool_args)
print(f"\n\n[Calling tool {tool_name} with args {tool_args}]\n\n")
# Store the data returned by DeepSeek about which tool was called and the data after the tool execution in messages
messages.append(content.message.model_dump())
messages.append({
"role": "tool",
"content": result.content[0].text,
"tool_call_id": tool_call.id,
})
# Return the above result to DeepSeek for generating the final result
response = self.client.chat.completions.create(
model=os.getenv("OPENAI_MODEL"),
messages=messages,
)
return response.choices[0].message.content
return content.message.content
```
Next, we implement a loop for querying and a cleanup operation to close the session after exiting.
```python
async def chat_loop(self):
while True:
try:
query = input("\nQuery: ").strip()
if query.lower() == 'quit':
break
response = await self.process_query(query)
print("\n" + response)
except Exception as e:
import traceback
traceback.print_exc()
async def cleanup(self):
"""Clean up resources"""
await self.exit_stack.aclose()
```
Finally, we complete the code related to running this client.
```python
async def main():
client = MCPClient()
try:
await client.connect_to_server()
await client.chat_loop()
finally:
await client.cleanup()
if __name__ == "__main__":
import sys
asyncio.run(main())
```
This is a minimal code example that does not implement message context recording and other features, simply to understand how to use a large language model to invoke the MCP server. This example only demonstrates how to connect to a single server. If you want to connect to multiple MCP servers, you can simply loop through the code in `connect_to_server`, encapsulate them into a class, and then iterate through all the tools in the MCP servers to generate a large `available_tools`, which can then be called based on the return results from the large language model. This will not be elaborated further here.
> You can refer to the official example: https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/clients/simple-chatbot/mcp_simple_chatbot/main.py
## Explanation of Sampling
MCP also provides us with a `Sampling` feature, which may seem confusing at first glance, but in fact, this feature gives us an interface to execute some operations before and after executing a tool. For example, when calling a tool to delete a local file, we would definitely want to confirm before proceeding with the deletion. In this case, we can use this feature.
Let's implement this supervised functionality.
First, we will create a simulated MCP server that has the ability to delete files:
```python
# Server
from mcp.server import FastMCP
from mcp.types import SamplingMessage, TextContent
app = FastMCP('file_server')
@app.tool()
async def delete_file(file_path: str):
# Create SamplingMessage to trigger the sampling callback function
result = await app.get_context().session.create_message(
messages=[
SamplingMessage(
role='user', content=TextContent(
type='text', text=f'Do you want to delete the file: {file_path}? (Y)')
)
],
max_tokens=100
)
# Get the return value from the sampling callback function and process it
if result.content.text == 'Y':
return f'File {file_path} has been deleted!!'
if __name__ == '__main__':
app.run(transport='stdio')
```
The most important part here is to create a `SamplingMessage` type message using the `create_message` method, which will send this message to the corresponding sampling callback function.
Next, we will create the client code:
```python
# Client
import asyncio
from mcp.client.stdio import stdio_client
from mcp import ClientSession, StdioServerParameters
from mcp.shared.context import RequestContext
from mcp.types import (
TextContent,
CreateMessageRequestParams,
CreateMessageResult,
)
server_params = StdioServerParameters(
command='uv',
args=['run', 'file_server.py'],
)
async def sampling_callback(
context: RequestContext[ClientSession, None],
params: CreateMessageRequestParams,
):
# Get the message sent by the tool and display it to the user
input_message = input(params.messages[0].content.text)
# Send the user input back to the tool
return CreateMessageResult(
role='user',
content=TextContent(
type='text',
text=input_message.strip().upper() or 'Y'
),
model='user-input',
stopReason='endTurn'
)
async def main():
async with stdio_client(server_params) as (stdio, write):
async with ClientSession(
stdio, write,
# Set the corresponding sampling_callback method
sampling_callback=sampling_callback
) as session:
await session.initialize()
res = await session.call_tool(
'delete_file',
{'file_path': 'C:/xxx.txt'}
)
# Get the final result returned by the tool
print(res)
if __name__ == '__main__':
asyncio.run(main())
```
It is important to note that currently, the content printed inside the tool cannot be displayed in the command line window when using `stdio_client`. Therefore, for debugging, we can use `mcp.shared.memory.create_connected_server_and_client_session`.
The specific code is as follows:
```python
# Client
from mcp.shared.memory import (
create_connected_server_and_client_session as create_session
)
# Here we need to import the server's app object
from file_server import app
async def sampling_callback(context, params):
...
async def main():
async with create_session(
app._mcp_server,
sampling_callback=sampling_callback
) as client_session:
await client_session.call_tool(
'delete_file',
{'file_path': 'C:/xxx.txt'}
)
if __name__ == '__main__':
asyncio.run(main())
```
## Loading MCP Server into Claude Desktop
The next two features are actually aimed at enhancing the experience for the Claude desktop client, so let's first discuss how to load our custom MCP server into the Claude desktop client.
First, we open the configuration.
We click on the `Developer` menu, then click the `Edit Config` button to open the Claude desktop client configuration file `claude_desktop_config.json`.
Next, we start adding our server. The server needs to be under the `mcpServers` hierarchy, with parameters including `command`, `args`, and `env`. In fact, the parameters are the same as those used to initialize the `StdioServerParameters` object.
```json
{
"mcpServers": {
"web-search-server": {
"command": "uv",
"args": [
"--directory",
"D:/projects/mcp_getting_started",
"run",
"web_search.py"
]
}
}
}
```
Finally, after saving the file and restarting the Claude desktop client, we can see our plugin here.
Of course, we can also directly run the following command in the directory of our plugin to install it:
```shell
mcp install web_search.py
```
## Other Features
### Prompt
MCP also provides us with a feature to generate Prompt templates. It is very simple to use; just decorate it with the `prompt` decorator, as shown in the code below:
```python
from mcp.server import FastMCP
app = FastMCP('prompt_and_resources')
@app.prompt('Translation Expert')
async def translate_expert(
target_language: str = 'Chinese',
) -> str:
return f'You are a translation expert, skilled at translating any language into {target_language}. Please translate the following content:'
if __name__ == '__main__':
app.run(transport='stdio')
```
Then we can add our new MCP server using the method mentioned in the previous section to configure the Claude desktop client. After that, we can click the icon in the lower right corner to start using it.
It will prompt us to set the parameters we pass in, and it will generate an attachment in our chat window.
### Resource
We can also allow users to select preset resources we provide on the Claude client, and it also supports custom protocols. The specific code is as follows:
```python
from mcp.server import FastMCP
app = FastMCP('prompt_and_resources')
@app.resource('echo://static')
async def echo_resource():
# The content returned when the user uses this resource
return 'Echo!'
@app.resource('greeting://{name}')
async def get_greeting(name):
return f'Hello, {name}!'
if __name__ == '__main__':
app.run(transport='stdio')
```
Then, we can check it on the Claude desktop client.
It is important to note that currently, the Claude desktop client cannot read resource decorators set with wildcard paths like `greeting://{name}`. This will be supported in the future. However, in our client code, it can be used as a resource template, as shown in the following code:
```python
import asyncio
from pydantic import AnyUrl
from mcp.client.stdio import stdio_client
from mcp import ClientSession, StdioServerParameters
server_params = StdioServerParameters(
command='uv',
args=['run', 'prompt_and_resources.py'],
)
async def main():
async with stdio_client(server_params) as (stdio, write):
async with ClientSession(stdio, write) as session:
await session.initialize()
# Get the list of resources without wildcards
res = await session.list_resources()
print(res)
# Get the list of resources with wildcards (resource templates)
res = await session.list_resource_templates()
print(res)
# Read resources, matching wildcards
res = await session.read_resource(AnyUrl('greeting://liming'))
print(res)
# Get the list of Prompt templates
res = await session.list_prompts()
print(res)
# Use Prompt templates
res = await session.get_prompt(
'Translation Expert', arguments={'target_language': 'English'})
print(res)
if __name__ == '__main__':
asyncio.run(main())
```
## Using MCP Server in LangChain
Recently, LangChain released a new open-source project called `langchain-mcp-adapters`, which makes it easy to integrate MCP servers into LangChain. Let's see how to use it:
```python
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
from langchain.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o")
server_params = StdioServerParameters(
command='uv',
args=['run', 'web_search.py'],
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# Get the list of tools
tools = await load_mcp_tools(session)
# Create and use ReAct agent
agent = create_react_agent(model, tools)
agent_response = await agent.ainvoke({'messages': 'What is the weather like in Hangzhou today?'})
```
For more detailed usage, please refer to: https://github.com/langchain-ai/langchain-mcp-adapters
## DeepSeek + cline + Custom MCP = Image and Text Master
Finally, let's use the cline plugin in VsCode to build an Image and Text Master application through DeepSeek and our custom image generation MCP server. Without further ado, let's get started.
First, we will build our image generation MCP server. Here, we will directly use the `FLUX.1-schnell` model from Hugging Face, available at: https://huggingface.co/spaces/black-forest-labs/FLUX.1-schnell. We will not use the `gradio_client` library but will instead use `httpx` to implement it manually, as using `gradio_client` may lead to encoding errors. The specific code is as follows:
```python
# image_server.py
import json
import httpx
from mcp.server import FastMCP
app = FastMCP('image_server')
@app.tool()
async def image_generation(image_prompt: str):
"""
Generate an image
:param image_prompt: Image description, must be in English
:return: Local path where the image is saved
"""
async with httpx.AsyncClient() as client:
data = {'data': [image_prompt, 0, True, 512, 512, 3]}
# Create a task to generate the image
response1 = await client.post(
'https://black-forest-labs-flux-1-schnell.hf.space/call/infer',
json=data,
headers={"Content-Type": "application/json"}
)
# Parse the response to get the event ID
response_data = response1.json()
event_id = response_data.get('event_id')
if not event_id:
return 'Unable to get event ID'
# Stream the response data
url = f'https://black-forest-labs-flux-1-schnell.hf.space/call/infer/{event_id}'
full_response = ''
async with client.stream('GET', url) as response2:
async for chunk in response2.aiter_text():
full_response += chunk
return json.loads(full_response.split('data: ')[-1])[0]['url']
if __name__ == '__main__':
app.run(transport='stdio')
```
Next, we can use the following command in the virtual environment to open the MCP Inspector and debug our tool.
```shell
mcp dev image_server.py
```
Then, we install the cline plugin in VsCode. After installing the plugin, we configure our DeepSeek API key. Next, we click the `MCP Server` button in the upper right corner to open the MCP server list.
Then switch to the `Installed` tab and click the `Configure MCP Servers` button to edit the custom MCP server.
The configuration is as follows:
```json
{
"mcpServers": {
"image_server": {
"command": "uv",
"args": [
"--directory",
"D:/projects/mcp_getting_started",
"run",
"image_server.py"
],
"env": {},
"disabled": false,
"autoApprove": []
}
}
}
```
After saving, if the small dot here turns green, it indicates that our server is connected, and we can start using it!
Next, we open the input box to enter the content of the article we want to write:
We can see that it correctly called our tool.
Finally, we can see the generated article.
Thus, the entire MCP introductory tutorial comes to an end. The related code will be available on GitHub: https://github.com/liaokongVFX/MCP-Chinese-Getting-Started-Guide
