Content
# Tool List
[TOC]
## Introduction
The Model Context Protocol (MCP) is an innovative open-source protocol that redefines how large language models (LLMs) interact with the external world. MCP provides a standardized method for any large language model to easily connect to various data sources and tools, enabling seamless information access and processing. MCP can be thought of as the USB-C interface for AI applications, providing a standardized way for AI models to connect to different data sources and tools.
MCP has the following core features:
- Resources
- Prompts
- Tools
- Sampling
- Roots
- Transports
Since most features are designed to serve the Claude client, this article focuses on writing the MCP server service for general large language models, with a focus on "tools" and placing other features at the end for a brief explanation.
The MCP transport layer supports two protocol implementations: stdio (standard input/output) and SSE (server-sent events). Since stdio is more commonly used, this article will use it as an example.
This article will use Python version 3.11 and uv to manage the Python project. The code will be placed on Github at the end.
## Developing the MCP Server
In this section, we will implement a server for web search. First, let's initialize our project using uv.
> uv Official Documentation: https://docs.astral.sh/uv/
```shell
# Initialize the project
uv init mcp_getting_started
cd mcp_getting_started
# Create a virtual environment and enter it
uv venv
.venv\Scripts\activate.bat
# Install dependencies
uv add "mcp[cli]" httpx openai
```
Then, let's create a file called `web_search.py` to implement our service. MCP provides two objects: `mcp.server.FastMCP` and `mcp.server.Server`. We will use `mcp.server.FastMCP` here, which is a higher-level encapsulation.
```python
import httpx
from mcp.server import FastMCP
# Initialize the FastMCP server
app = FastMCP('web-search')
```
Implementing the execution method is simple. MCP provides a `@mcp.tool()` decorator, and we just need to decorate the implementation function with it. The function name will be used as the tool name, and the parameters will be used as tool parameters, described by comments, along with the return value.
Here, we directly use the interface provided by Zhi Pu, which not only helps us search for relevant result links but also generates summaries of the content of the corresponding links. ~~And at this stage, it is free~~ (it has started charging, 0.03 yuan per request), making it very suitable for us.
> 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 for internet content
Args:
query: The content to search for
Returns:
A summary of the 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 writing of the MCP server. Next, let's use the official `Inspector` visualization tool to debug our server.
We can run `Inspector` in two ways:
> Make sure you have installed the Node environment.
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:
```shell
mcp dev PYTHONFILE
```
The command to run our code is:
```shell
mcp dev web_search.py
```
When the following prompt appears, it means the operation is successful. If the connection fails, it may be due to port occupation. You can refer to this issue for a solution: https://github.com/liaokongVFX/MCP-Chinese-Getting-Started-Guide/issues/6
Then, we open this address, click the `Connect` button on the left, and connect to our service. We switch to the `Tools` column, click the `List Tools` button, and see our tool. We can start debugging now.
## Developing the MCP Client
First, let's see how to call the tools in the MCP server we just developed on the client.
```python
import asyncio
from mcp.client.stdio import stdio_client
from mcp import ClientSession, StdioServerParameters
```
# Tool List
## Creating Server Parameters for stdio Connection
server_params = StdioServerParameters(
# Command to be executed by the server, here we use uv to run web_search.py
command='uv',
# Arguments to be passed to the command
args=['run', 'web_search.py'],
# Environment variables, default is None, which means using the current environment variables
# env=None
)
async def main():
# Create a stdio client
async with stdio_client(server_params) as (stdio, write):
# Create a ClientSession object
async with ClientSession(stdio, write) as session:
# Initialize the ClientSession
await session.initialize()
# List available tools
response = await session.list_tools()
print(response)
# Call a 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())
## Explanation
Our Python script needs to be run in a virtual environment, so we use `uv` to start our script.
## Example
Let's see how to use `DeepSeek` to call methods in our MCP server.
We will use `dotenv` to manage our environment variables. The contents of the `.env` file are as follows:
```shell
OPENAI_API_KEY=sk-89baxxxxxxxxxxxxxxxxxx
OPENAI_BASE_URL=https://api.deepseek.com
OPENAI_MODEL=deepseek-chat
```
## Implementation
First, let's implement 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()
```
## Adding connect_to_server Method
```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()
```
## Implementing process_query Method
```python
async def process_query(self, query: str) -> str:
# Here we need to constrain the large language model with system prompt,
# otherwise, it may not call tools and answer on its own.
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 tool list information from the MCP server
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, pass function call description information 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 tool is needed, parse the tool
tool_calls = content.message.tool_calls[0]
tool_name = tool_calls.function.name
tool_args = json.loads(tool_calls.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 tool call and execution result in messages
messages.append(content.message.model_dump())
messages.append({
"role": "tool",
"content": result.content[0].text,
"tool_calls_id": tool_calls.id,
})
# Return the result to DeepSeek for final result generation
response = self.client.chat.completions.create(
model=os.getenv("OPENAI_MODEL"),
messages=messages,
)
return response.choices[0].message.content
return content.message.content
```
## Implementing chat_loop and cleanup Methods
```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()
```
## Running the 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())
```
## Sampling Explanation
MCP provides a `Sampling` feature, which allows us to execute operations before and after tool execution. For example, when calling a tool to delete a local file, we may want to confirm with the user before deletion.
## Implementing Sampling
First, let's create a simulated MCP server with a delete file tool:
```python
# Server side
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 a 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'Are you sure to delete the file: {file_path} (Y)')
)
],
max_tokens=100
)
# Get the sampling callback function's return value and process accordingly
if result.content.text == 'Y':
return f'The file {file_path} has been deleted!!'
if __name__ == '__main__':
app.run(transport='stdio')
```
## Client Side Implementation
```python
# Client side
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 tool's message and display it to the user
input_message = input(params.messages[0].content.text)
# Send the user's 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 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 tool's final execution result
print(res)
if __name__ == '__main__':
asyncio.run(main())
```
## Tool List
## Loading MCP Server on Claude Desktop
Because the next two functions are actually provided for Claude desktop, let's talk about how to load our custom MCP Server to Claude desktop.
First, let's open the configuration.
We click the `Developer` menu and then click the `Edit Config` button to open the configuration file `claude_desktop_config.json` for Claude desktop.
Then, we start adding our server. The server needs to be under the `mcpServers` level, with parameters `command`, `args`, and `env`. In fact, the parameters are the same as those when initializing the `StdioServerParameters` object.
```json
{
"mcpServers": {
"web-search-server": {
"command": "uv",
"args": [
"--directory",
"D:/projects/mcp_getting_started",
"run",
"web_search.py"
]
}
}
}
```
Finally, we save the file and restart Claude desktop to see our plugin here.
Of course, we can also directly run the following command in our plugin directory to install:
```shell
mcp install web_search.py
```
## Other Functions
### Prompt
MCP also provides a function to generate Prompt templates. It's easy to use, just decorate it with the `prompt` decorator, as shown in the code:
```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, good at translating any language into {target_language}. Please translate the following content:'
if __name__ == '__main__':
app.run(transport='stdio')
```
Then, we use the method mentioned in the previous section to configure the Claude desktop MCP server, and then we can click the icon in the bottom right corner to start using it.
It will let us set the parameters we pass in, and then it will generate an attachment in our chat window.
### Resource
We can also select the preset resources provided for users on the Claude client, and also support 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():
# Return when the user uses this resource, the content of the resource
return 'Echo!'
@app.resource('greeting://{name}')
async def get_greeting(name):
return f'Hello, {name}!'
if __name__ == '__main__':
app.run(transport='stdio')
```
Then, let's take a look at the Claude desktop.
Here, we need to pay special attention to the fact that the current Claude desktop can't read the resource decorator setting `greeting://{name}` this kind of wildcard path, and it will be supported in the future. However, in our client code, it can be used as a resource template, the specific code is as follows:
```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, will match 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())
```
### Lifecycle
The MCP lifecycle is divided into three stages:
- Initialization
- Interaction communication
- Service is closed
Therefore, we can do some things at the beginning and end of these three stages, such as creating and closing database connections, recording logs, recording tool usage information, etc.
Below, we will take the web search tool as an example, and store the query and query results in a global context as a cache. The complete code is as follows:
```python
import httpx
from dataclasses import dataclass
from contextlib import asynccontextmanager
from mcp.server import FastMCP
from mcp.server.fastmcp import Context
@dataclass
# Initialize a lifecycle context object
class AppContext:
# There is a field to store request history
histories: dict
@asynccontextmanager
async def app_lifespan(server):
# Execute when MCP initializes
histories = {}
try:
# Each communication will pass this context as a parameter to the tool
yield AppContext(histories=histories)
finally:
# Execute when MCP service is closed
print(histories)
app = FastMCP(
'web-search',
# Set lifecycle listener function
lifespan=app_lifespan
)
@app.tool()
# The first parameter will be passed to the context object
async def web_search(ctx: Context, query: str) -> str:
"""
Search Internet content
Args:
query: Content to search
Returns:
Summary of search results
"""
# If the same question has been asked before, directly return the cache
histories = ctx.request_context.lifespan_context.histories
if query in histories:
return histories[query]
async with httpx.AsyncClient() as client:
response = await client.post(
'https://open.bigmodel.cn/api/paas/v4/tools',
headers={'Authorization': 'YOUR 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_data = '\n\n\n'.join(res_data)
# Store query value and return value in histories
ctx.request_context.lifespan_context.histories[query] = return_data
return return_data
if __name__ == "__main__":
app.run()
```
## Using MCP Server in LangChain
Recently, LangChain released a new open-source project `langchain-mcp-adapters`, which can easily integrate MCP servers into LangChain. Let's take a look at 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 langgraph.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 tool list
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 = Graphic Master
Finally, let's use the cline plugin of VsCode to build a graphic master application through DeepSeek and a custom MCP image generation server. Let's get started.
First, let's build our image generation MCP server. Here, we directly use the `FLUX.1-schnell` model on huggingface: https://huggingface.co/spaces/black-forest-labs/FLUX.1-schnell. Here, we don't use the `gradio_client` library, but instead, we use `httpx` to create a server. 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 image
:param image_prompt: Image description, needs to 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 image generation task
response1 = await client.post(
'https://black-forest-labs-flux-1-schnell.hf.space/call/infer',
json=data,
headers={"Content-Type": "application/json"}
)
# Parse response to get event ID
response_data = response1.json()
event_id = response_data.get('event_id')
if not event_id:
return 'Unable to get event ID'
# Get return data through streaming
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')
```
Then, 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
```
Next, we install the cline plugin in VsCode. After installing the plugin, we configure our DeepSeek API key. Then, we click the `MCP Server` button in the top 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, the small dot here is green, indicating that our server is connected, and then we can start using it.
Then, we open the input box and enter the content of the article we want to write:
We can see that it correctly calls our tool
Finally, we can see the generated article.
## Deploying MCP Services to the Cloud with Serverless
So far, we've discussed how to use local MCP services, but sometimes we want to deploy MCP services directly to the cloud for direct invocation, eliminating the need for local downloads and startup. To achieve this, we need to use MCP's SSE protocol.
First, let's write an MCP service using the SSE protocol. This is quite simple; we just need to set the `transport` parameter in the final `run` command to `sse`. Below, we'll use the web search example from earlier to implement this, with the specific code as follows:
```python
# sse_web_search.py
import httpx
from mcp.server import FastMCP
app = FastMCP('web-search', port=9000)
@app.tool()
async def web_search(query: str) -> str:
"""
Search for 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': 'YOUR 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)
if __name__ == "__main__":
app.run(transport='sse')
```
In `FastMCP`, there are several parameters related to the SSE protocol that can be set:
- host: Service address, defaults to `0.0.0.0`
- port: Service port, defaults to 8000. In the code above, it's set to `9000`
- sse_path: SSE route, defaults to `/sse`
Now, we can write a client code to test it. The specific code is as follows:
```python
import asyncio
from mcp.client.sse import sse_client
from mcp import ClientSession
async def main():
async with sse_client('http://localhost:9000/sse') as streams:
async with ClientSession(*streams) as session:
await session.initialize()
res = await session.call_tool('web_search', {'query': '杭州今天天气'})
print(res)
if __name__ == '__main__':
asyncio.run(main())
```
We can see that it works normally and searches for content:
Of course, we can also use `mcp dev sse_web_search.py` to test. Here, note that the `Transport Type` needs to be changed to `SSE`, and then fill in our local service address.
When everything is tested without issues, we'll deploy it to the cloud using Serverless. Here, we choose Alibaba Cloud's Function Compute service. First, we enter the `Functions` menu of Alibaba Cloud's `Function Compute FC 3.0` and click `Create Function` to create our service. The address is: https://fcnext.console.aliyun.com/cn-hangzhou/functions
We choose `Web Function` here, and for the runtime environment, we select `Python 10`. The code upload method can be chosen according to your needs; since there's only one Python file here, we choose `Use example code` directly, so we can put our code in later. The startup command and listening port are retained as default (**the port needs to be consistent with the code**).
Environment variables can be set to API keys used in the code; here, we don't set them. The final setup screenshot is as follows:
In advanced settings, for easy debugging, we enable the log function.
After setting up, click Create. It will jump to the code editing section, and we can paste our previous code in.
Next, we install dependencies. Click the `Edit Layer` in the top right corner. There will be a default Flask layer; since the template uses Flask, we don't need it. We delete it and add an MCP layer. Choose `Add Official Public Layer`, search for `mcp`, and there will be a Python version of the MCP layer containing all MCP dependencies.
If you have other third-party dependencies, you can search for them in the public layer first; if not available, you can build a custom layer. Click here to do so; you just need to provide a `requirements` list. This is not elaborated on here.
When everything is set up, click Deploy in the bottom right corner.
Then, we go back to the code editing page and click Deploy Code in the top left corner. After a few seconds, it will indicate that the code deployment is successful. Now, our MCP service is deployed to the cloud.
> Update on 20250409: Not sure if the official noticed this article, but now you can directly choose `MCP Runtime` when running, eliminating the need to manually add an `MCP Layer` in the layer section.
>
> 
Then, we switch to the `Triggers` section under `Configurations` and can see the URL address for access. Of course, you can also bind your own domain.
Now, we can test using the client code above.
```python
import asyncio
from mcp.client.sse import sse_client
from mcp import ClientSession
async def main():
async with sse_client('https://mcp-test-whhergsbso.cn-hangzhou.fcapp.run/sse') as streams:
async with ClientSession(*streams) as session:
await session.initialize()
res = await session.call_tool('web_search', {'query': '杭州今天天气'})
print(res)
if __name__ == '__main__':
asyncio.run(main())
```
If we encounter errors in the client, don't panic; we can directly find the corresponding request in the logs, click `Request Log`, and view the error to fix it.
By now, our MCP service is deployed to the cloud, and we can use it directly from anywhere.
For example, in `Cherry-Studio`, we can set it like this:
In `Cline`:
In `Cursor`:
```json
{
"mcpServers": {
"web-search": {
"url": "https://mcp-test-whhergsbso.cn-hangzhou.fcapp.run/sse"
}
}
}
```
That's it for the MCP getting started tutorial. Subsequent updates will be made if there are other relevant topics. The related code will be placed in the GitHub repository: https://github.com/liaokongVFX/MCP-Chinese-Getting-Started-Guide
