MCP 是一種開放協議，通過標準化的服務器實現，使 AI 模型能夠安全地與本地和遠程資源進行交互。MCP 可幫助你在 LLM 之上構建智能代理和復雜的工作流。MCP 采用客戶端-服務器架構，主機應用程序可以連接到多個服務器。
這里用個demo展示一下如何使用MCP。首先需要pip install mcp安裝庫。

1. 服務端

服務端用@mcp.tool注冊服務即可：

import json,httpx,requests,logging,sys
from typing import Any
import pandas as pd
from pandasql import sqldf
from mcp.server.fastmcp  import FastMCP
import warnings
from qwen_agent.llm import get_chat_model
from dotenv import load_dotenv
logger = logging.getLogger(__name__)
load_dotenv()
warnings.filterwarnings("ignore")
mcp = FastMCP("Server")
bearer = os.getenv('BEARER')
llm = get_chat_model({"model": os.getenv('MODEL_NAME'),"model_server": os.getenv('MODEL_SERVER'),"api_key":os.getenv('MODEL_KEY')})@mcp.tool()
def query_hour(question: str, start_at: str="2025-04-01", end_at: str="2025-06-01") -> str:"""輸入關于時間范圍相關問題，返回問題的查詢結果。:param question: 關于時間范圍的問題:param start_at: 查詢開始時間，時間格式為"2025-01-01":param end_at: 查詢結束時間，時間格式為"2025-06-01":return: 問題的查詢結果"""res = requests.post("***",json = {"start_at": start_at,"end_at": end_at},verify=False,headers ={"Authorization":bearer}).json()for response in llm.chat([{"role": "system", "content": f"結果信息如下{str(res)}，請盡量簡潔回復問題"},{"role": "user", "content": f"{question}"}]):passreturn response[0]['content']if __name__ == "__main__":# 以標準 I/O 方式運行 MCP 服務器logging.basicConfig(stream=sys.stdout, level=logging.INFO)logger.info('start server')mcp.run(transport='stdio')

2. 客戶端

客戶端可以用任意的mcp前端，這里手動擼一個：

import asyncio,logging,os,json
from typing import Optional
from contextlib import AsyncExitStack
from qwen_agent.llm import get_chat_model
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from dotenv import load_dotenv
load_dotenv()
class MCPClient:def __init__(self):"""初始化 MCP 客戶端"""self.exit_stack = AsyncExitStack()self.llm = get_chat_model({"model": os.getenv('MODEL_NAME')',"model_server": os.getenv('MODEL_SERVER')',,"api_key":'MODEL_KEY'})self.session: Optional[ClientSession] = Noneself.exit_stack = AsyncExitStack()        async def connect_to_server(self, server_script_path: str):"""連接到 MCP 服務器并列出可用工具"""is_python = server_script_path.endswith('.py')is_js = server_script_path.endswith('.js')if not (is_python or is_js):raise ValueError("服務器腳本必須是 .py 或 .js 文件")command = "python" if is_python else "node"server_params = StdioServerParameters(command=command,args=[server_script_path],env=None)# 啟動 MCP 服務器并建立通信stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))self.stdio, self.write = stdio_transportself.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))await self.session.initialize()# 列出 MCP 服務器上的工具response = await self.session.list_tools()print("\n已連接到服務器，支持以下工具:", [tool.name for tool in response.tools])     async def process_query(self, query: str) -> str:"""使用大模型處理查詢并調用可用的MCP工具 (Function Calling)"""messages = [{"role": "user", "content": query}]tool_response = await self.session.list_tools()functions = [{"name": tool.name,"description": tool.description,"parameters": tool.inputSchema} for tool in tool_response.tools]for responses in self.llm.chat(messages=messages,functions=functions):passfor message in responses:if fn_call := message.get("function_call", None):fn_name: str = fn_call['name']fn_args: dict = json.loads(fn_call["arguments"])result = await self.session.call_tool(fn_name, fn_args)fn_res: str = result.content[0].textmessages.append({"role": "function","name": fn_name,"content": fn_res,})# 將上面的結果再返回給大模型用于生產最終的結果for responses in self.llm.chat(messages=messages, functions=functions):passreturn responses[0]['content'] async def chat_loop(self):"""運行交互式聊天循環"""print("\n🤖 MCP 客戶端已啟動！輸入 'quit' 退出")while True:try:query = input("\n你: ").strip()if query.lower() == 'quit':breakresponse = await self.process_query(query)  # 發送用戶輸入到APIif '錯誤' in response:response = await self.process_query(response+query)print(f"\n🤖 LighthouseAI: {response}")except Exception as e:print(f"\n?? 發生錯誤: {str(e)}")async def cleanup(self):"""清理資源"""await self.exit_stack.aclose()async def main():if len(sys.argv) < 2:print("Usage: python client.py <path_to_server_script>")sys.exit(1)client = MCPClient()try:await client.connect_to_server(sys.argv[1])await client.chat_loop()finally:await client.cleanup()if __name__ == "__main__":import syslogging.basicConfig(stream=sys.stdout, level=logging.INFO)asyncio.run(main())

3. 使用方式

啟動MCP clinet： uv run client.py server.py
會彈出一個聊天界面，可以在后面輸入問題，mcp服務器會根據問題找到相對應的tool進行回復：