MCP協(xié)議從原理到開發(fā)：一文讀懂大模型交互的標(biāo)準(zhǔn)化革命！

import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';export const server = new McpServer({  name: 'wechat-mcp-server',  version: '0.0.1',});

import { sendWeChatTextMessage } from './text.js';// Register WeChat message toolserver.tool(  'sendWeChatTextMessage',  'send WeChat text message',  {      webhookKey: z.string().describe('WeChat webhook key'), // 必填      content: z.string().describe('WeChat message content'),// 必填      chatid: z.string().describe('WeChat message chatid').optional(), // optional 選填      mentioned_list: z.array(z.string()).describe('WeChat message mentioned list').optional(),      mentioned_mobile_list: z.array(z.string()).describe('WeChat message mentioned mobile list').optional(),  },    async ({ webhookKey, content, chatid, mentioned_list, mentioned_mobile_list }) => {        const success = await sendWeChatTextMessage(webhookKey, content, chatid, mentioned_list,    mentioned_mobile_list);           if (!success) {          return {                content: [                          {                              type: 'text',                               text: 'Failed to send WeChat message',                    },               ],           };       }          return {            content: [                {                    type: 'text',                    text: 'WeChat message sent successfully',                 },           ],         };       },    );

server.prompt(  'parameter_check',  '參數(shù)檢查',  {     param: z.string().describe('參數(shù)'),    apiContent: z.string().describe('API文檔')  },  ({ param,apiContent }) => ({      messages: [{          role: 'user',          content: {              type: 'text',              text: `              <instruction>                <instructions>                    1. 接收API文檔文本和用戶數(shù)據(jù)作為輸入。                    2. 從API文檔中提取所有參數(shù)的要求，包括參數(shù)名稱、類型、是否必需、默認(rèn)值等信息。                    3. 對照用戶提供的數(shù)據(jù)，檢查每個參數(shù)是否滿足API的要求。                    4. 對于每個參數(shù)，記錄以下信息：                        - 參數(shù)名稱                        - 用戶提供的值                        - 是否滿足API要求（是/否）                        - 如果不滿足，說明原因                    5. 將所有檢查結(jié)果整理成一個清晰的報告，確保輸出不包含任何XML標(biāo)簽。                    6. 確保報告的格式簡潔明了，便于用戶理解。                </instructions>                <examples>                   <example>                      <input>                          API文檔文本: '參數(shù): username, 類型: string, 必需: 是; 參數(shù): age, 類型: integer, 必需: 否;'                          用戶數(shù)據(jù): '{\'username\': john_doe, \'age\': 25}'                      </input>                      <output>                          '參數(shù): username, 用戶提供的值: john_doe, 滿足要求: 是; 參數(shù): age, 用戶提供的值: 25, 滿足要求: 是;'                      </output>                  </example>                </examples>              </instruction>              API文檔文本：${apiContent}              用戶數(shù)據(jù)：${param}              `          }      }]  }));

async def connect_to_sse_server(self, server_url: str):    '''連接到使用 SSE 傳輸?shù)?MCP 服務(wù)器'''    if not server_url.startswith(' http://' ) and not server_url.startswith(' https://' ):        raise ValueError('服務(wù)器 URL 必須以 ' http://'  或 'https://' 開頭')
    # 存儲上下文管理器以保持其存活    self._streams_context = sse_client(url=server_url)    streams = await self._streams_context.__aenter__()
    self._session_context = ClientSession(*streams)    self.session: ClientSession = await self._session_context.__aenter__()
    # 初始化    await self.session.initialize()    async def connect_to_server(self, server_script_path: str):        '''        連接到通過標(biāo)準(zhǔn)輸入/輸出通信的 MCP 服務(wù)器                參數(shù):            server_script_path: 服務(wù)器腳本路徑(.py 或 .js 文件)                示例: 'server.py' 或 'server.js'        '''        is_python = server_script_path.endswith('.py')        is_js = server_script_path.endswith('.js')        if not (is_python or is_js):            raise ValueError('Server script must be a .py or .js file')
        command = 'python' if is_python else 'node'        server_params = StdioServerParameters(            command=command,            args=[server_script_path],            env=None        )
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))        self.stdio, self.write = stdio_transport        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
        await self.session.initialize()

async def main():    if len(sys.argv) < 2:        print('用法: uv run client.py <SSE MCP 服務(wù)器的 URL (例如 ' '=''> http://localhost:8080/sse)>' )        print('或: uv run client.py <MCP 服務(wù)器腳本路徑 (例如 server.py)>')        sys.exit(1)
    client = MCPClient()    try:        # stdio 連接方式        # await client.connect_to_server(server_script_path=sys.argv[1])        # sse 連接方式        await client.connect_to_sse_server(server_url=sys.argv[1])        print('\n連接到服務(wù)器...')                        response = await client.session.list_prompts()        response1 = await client.session.list_tools()        response2 = await client.session.list_resources()        print('response', response)        print('response1', response1)        print('response2', response2)        
    finally:        try:            await client.cleanup()        except Exception as e:            print(f'清理錯誤: {e}')

async def process_llm_response(self, llm_response: str) -> str:        '''Process the LLM response and execute tools if needed.
        Args:            llm_response: The response from the LLM.
        Returns:            The result of tool execution or the original response.        '''        import json
        try:            tool_call = json.loads(llm_response)            #如果有工具被選中，開始執(zhí)行工具            if 'tool' in tool_call and 'arguments' in tool_call:                logging.info(f'Executing tool: {tool_call['tool']}')                logging.info(f'With arguments: {tool_call['arguments']}')                #這里的server是通過mcp client 發(fā)現(xiàn)的                for server in self.servers:                    tools = await server.list_tools()                    if any(tool.name == tool_call['tool'] for tool in tools):                        try:                            # 執(zhí)行工具                            result = await server.execute_tool(                                tool_call['tool'], tool_call['arguments']                            )
                            #... 省略了無關(guān)緊要的代碼                            return f'Tool execution result: {result}'                        except Exception as e:                            error_msg = f'Error executing tool: {str(e)}'                            logging.error(error_msg)                            return error_msg
                return f'No server found with tool: {tool_call['tool']}'            # 沒有工具執(zhí)行就直接輸出模型的回答            return llm_response        except json.JSONDecodeError:            return llm_response