### Basic MCP Client StdIO Connection (Python) Source: https://github.com/liaokongvfx/mcp-chinese-getting-started-guide/blob/main/README.md Demonstrates a simple, standalone example of connecting to an MCP server via stdio, initializing a client session, listing available tools, and calling a specific tool (`web_search`). Uses `asyncio` for asynchronous operations. ```Python server_params = StdioServerParameters( # 服务器执行的命令，这里我们使用 uv 来运行 web_search.py command='uv', # 运行的参数 args=['run', 'web_search.py'], # 环境变量，默认为 None，表示使用当前环境变量 # env=None ) async def main(): # 创建 stdio 客户端 async with stdio_client(server_params) as (stdio, write): # 创建 ClientSession 对象 async with ClientSession(stdio, write) as session: # 初始化 ClientSession await session.initialize() # 列出可用的工具 response = await session.list_tools() print(response) # 调用工具 response = await session.call_tool('web_search', {'query': '今天杭州天气'}) print(response) if __name__ == '__main__': asyncio.run(main()) ``` -------------------------------- ### Defining MCP Resources in Python Source: https://github.com/liaokongvfx/mcp-chinese-getting-started-guide/blob/main/README.md This snippet shows how to define resources using the `@app.resource` decorator in MCP. It includes examples for a static resource (`echo://static`) and a resource template with a wildcard parameter (`greeting://{name}`). Resources provide predefined content or dynamic responses based on the requested path. ```python from mcp.server import FastMCP app = FastMCP('prompt_and_resources') @app.resource('echo://static') async def echo_resource(): # 返回的是，当用户使用这个资源时，资源的内容 return 'Echo!' @app.resource('greeting://{name}') async def get_greeting(name): return f'Hello, {name}!' if __name__ == '__main__': app.run(transport='stdio') ``` -------------------------------- ### Configure MCP Server URL - JSON Source: https://github.com/liaokongvfx/mcp-chinese-getting-started-guide/blob/main/README.md Provides a JSON configuration example, likely for a client application or environment, specifying the URL for a deployed MCP server named 'web-search'. This snippet shows how to point a client to the public endpoint of the server running on Alibaba Cloud Function Compute. ```json { "mcpServers": { "web-search": { "url": "https://mcp-test-whhergsbso.cn-hangzhou.fcapp.run/sse" } } } ``` -------------------------------- ### Initialize MCPClient Class (Python) Source: https://github.com/liaokongvfx/mcp-chinese-getting-started-guide/blob/main/README.md Defines the basic structure of the `MCPClient` class, including necessary imports, loading environment variables using `dotenv`, and initializing instance attributes like the MCP session, an `AsyncExitStack` for managing asynchronous contexts, and an `OpenAI` client instance. ```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() ``` -------------------------------- ### MCP Client Interaction with Prompts and Resources in Python Source: https://github.com/liaokongvfx/mcp-chinese-getting-started-guide/blob/main/README.md This Python client code demonstrates how to interact with an MCP server to discover and utilize defined prompts and resources. It shows how to list available resources and prompt templates, read a specific resource (including one matching a template), and retrieve/use a prompt template with arguments. ```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() # 获取无通配符的资源列表 res = await session.list_resources() print(res) # 获取有通配符的资源列表(资源模板) res = await session.list_resource_templates() print(res) # 读取资源，会匹配通配符 res = await session.read_resource(AnyUrl('greeting://liming')) print(res) # 获取 Prompt 模板列表 res = await session.list_prompts() print(res) # 使用 Prompt 模板 res = await session.get_prompt( '翻译专家', arguments={'target_language': '英语'}) print(res) if __name__ == '__main__': asyncio.run(main()) ``` -------------------------------- ### Process Query with DeepSeek and MCP Tools (Python) Source: https://github.com/liaokongvfx/mcp-chinese-getting-started-guide/blob/main/README.md Implements the `process_query` method, which handles the interaction between a user query, the DeepSeek LLM, and the MCP server tools. It constructs messages with a system prompt, retrieves available MCP tools, calls the DeepSeek API with tool descriptions, processes the response (either direct content or tool calls), executes the tool via the MCP session if needed, and formats the final response from DeepSeek. ```Python async def process_query(self, query: str) -> str: # 这里需要通过 system prompt 来约束一下大语言模型， # 否则会出现不调用工具，自己乱回答的情况 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} ] # 获取所有 mcp 服务器 工具列表信息 response = await self.session.list_tools() # 生成 function call 的描述信息 available_tools = [{ "type": "function", "function": { "name": tool.name, "description": tool.description, "input_schema": tool.inputSchema } } for tool in response.tools] # 请求 deepseek，function call 的描述信息通过 tools 参数传入 response = self.client.chat.completions.create( model=os.getenv("OPENAI_MODEL"), messages=messages, tools=available_tools ) # 处理返回的内容 content = response.choices[0] if content.finish_reason == "tool_calls": # 如何是需要使用工具，就解析工具 tool_call = content.message.tool_calls[0] tool_name = tool_call.function.name tool_args = json.loads(tool_call.function.arguments) # 执行工具 result = await self.session.call_tool(tool_name, tool_args) print(f"\n\n[Calling tool {tool_name} with args {tool_args}]\n\n") # 将 deepseek 返回的调用哪个工具数据和工具执行完成后的数据都存入messages中 messages.append(content.message.model_dump()) messages.append({ "role": "tool", "content": result.content[0].text, "tool_call_id": tool_call.id, }) # 将上面的结果再返回给 deepseek 用于生产最终的结果 response = self.client.chat.completions.create( model=os.getenv("OPENAI_MODEL"), messages=messages, ) return response.choices[0].message.content return content.message.content ``` -------------------------------- ### Implementing MCP Lifecycle and Contextual Tool in Python Source: https://github.com/liaokongvfx/mcp-chinese-getting-started-guide/blob/main/README.md This snippet illustrates how to use the MCP lifecycle (`lifespan`) to manage application context, such as setting up a cache. It defines a `web_search` tool that accesses this context to implement simple caching for search queries, demonstrating how context can be passed to tools during interactions. ```python import httpx from dataclasses import dataclass from contextlib import asynccontextmanager from mcp.server import FastMCP from mcp.server.fastmcp import Context @dataclass # 初始化一个生命周期上下文对象 class AppContext: # 里面有一个字段用于存储请求历史 histories: dict @asynccontextmanager async def app_lifespan(server): # 在 MCP 初始化时执行 histories = {} try: # 每次通信会把这个上下文通过参数传入工具 yield AppContext(histories=histories) finally: # 当 MCP 服务关闭时执行 print(histories) app = FastMCP( 'web-search', # 设置生命周期监听函数 lifespan=app_lifespan ) @app.tool() # 第一个参数会被传入上下文对象 async def web_search(ctx: Context, query: str) -> str: """ 搜索互联网内容 Args: query: 要搜索内容 Returns: 搜索结果的总结 """ # 如果之前问过同样的问题，就直接返回缓存 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) # 将查询值和返回值存入到 histories 中 ctx.request_context.lifespan_context.histories[query] = return_data return return_data if __name__ == "__main__": app.run() ``` -------------------------------- ### Connect MCPClient to StdIO Server (Python) Source: https://github.com/liaokongvfx/mcp-chinese-getting-started-guide/blob/main/README.md Implements the `connect_to_server` method within the `MCPClient` class. This method configures the stdio server parameters, establishes the connection using `stdio_client`, enters the connection and client session into the `AsyncExitStack`, and initializes the client 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() ``` -------------------------------- ### Create MCP SSE Client and Call Tool - Python Source: https://github.com/liaokongvfx/mcp-chinese-getting-started-guide/blob/main/README.md Demonstrates how to create an asynchronous client to connect to an MCP server running with SSE transport. It uses 'sse_client' to establish the connection, initializes a 'ClientSession', calls the 'web_search' tool with a specific query ('杭州今天天气'), and prints the result received from the server. This snippet shows how to interact with the deployed or locally running SSE service. ```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()) ``` -------------------------------- ### Interactive Chat Loop (Python) Source: https://github.com/liaokongvfx/mcp-chinese-getting-started-guide/blob/main/README.md Implements the `chat_loop` method, providing a simple command-line interface for interacting with the `MCPClient`. It continuously prompts the user for input, processes the query using the `process_query` method, prints the response, and allows the user to exit by typing 'quit'. ```Python async def chat_loop(self): while True: try: query = input("\nQuery: ").strip() if query.lower() == 'quit': break response = await self.process_query(query) ``` -------------------------------- ### Define MCP SSE Server and Web Search Tool - Python Source: https://github.com/liaokongvfx/mcp-chinese-getting-started-guide/blob/main/README.md Initializes a FastMCP server instance configured for SSE transport on port 9000. It defines an asynchronous tool function 'web_search' that takes a query string, makes an HTTP POST request to an external web search API using 'httpx', processes the JSON response to extract search results, and returns a concatenated string of the results. The main block shows how to run the server with SSE. ```python import httpx from mcp.server import FastMCP app = FastMCP('web-search', port=9000) @app.tool() async def web_search(query: str) -> str: """ 搜索互联网内容 Args: query: 要搜索内容 Returns: 搜索结果的总结 """ 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') ``` -------------------------------- ### DeepSeek API Environment Variables (.env) Source: https://github.com/liaokongvfx/mcp-chinese-getting-started-guide/blob/main/README.md Shows the content of a `.env` file used to configure environment variables required for interacting with the DeepSeek API, including the API key, base URL, and model name. ```Shell OPENAI_API_KEY=sk-89baxxxxxxxxxxxxxxxxxx OPENAI_BASE_URL=https://api.deepseek.com OPENAI_MODEL=deepseek-chat ``` -------------------------------- ### Defining MCP Prompt Template in Python Source: https://github.com/liaokongvfx/mcp-chinese-getting-started-guide/blob/main/README.md This snippet demonstrates how to define a prompt template using the `@app.prompt` decorator in MCP. The `translate_expert` function takes a `target_language` argument and returns a formatted string intended for use as a prompt, specializing in translation. ```python from mcp.server import FastMCP app = FastMCP('prompt_and_resources') @app.prompt('翻译专家') async def translate_expert( target_language: str = 'Chinese', ) -> str: return f'你是一个翻译专家，擅长将任何语言翻译成{target_language}。请翻译以下内容：' if __name__ == '__main__': app.run(transport='stdio') ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.