多快好省，Qwen3混合部署模式引爆MCP

... # 省略了无关的代码async def start(self):     # 初始化所有的 mcp server     for server in self.servers:         await server.initialize()
     # 获取所有的 tools 命名为 all_tools     all_tools = []     for server in self.servers:         tools = await server.list_tools()         all_tools.extend(tools)
     # 将所有的 tools 的功能描述格式化成字符串供 LLM 使用     # tool.format_for_llm() 我放到了这段代码最后，方便阅读。     tools_description = "\n".join(         [tool.format_for_llm() for tool in all_tools]     )
     # 这里就不简化了，以供参考，实际上就是基于 prompt 和当前所有工具的信息     # 询问 LLM（Claude） 应该使用哪些工具。     system_message = (         "You are a helpful assistant with access to these tools:\n\n"         f"{tools_description}\n"         "Choose the appropriate tool based on the user's question. "         "If no tool is needed, reply directly.\n\n"         "IMPORTANT: When you need to use a tool, you must ONLY respond with "         "the exact JSON object format below, nothing else:\n"         "{\n"         '    "tool": "tool-name",\n'         '    "arguments": {\n'         '        "argument-name": "value"\n'         "    }\n"         "}\n\n"         "After receiving a tool's response:\n"         "1. Transform the raw data into a natural, conversational response\n"         "2. Keep responses concise but informative\n"         "3. Focus on the most relevant information\n"         "4. Use appropriate context from the user's question\n"         "5. Avoid simply repeating the raw data\n\n"         "Please use only the tools that are explicitly defined above."     )     messages = [{"role": "system", "content": system_message}]
     while True:         # Final... 假设这里已经处理了用户消息输入.         messages.append({"role": "user", "content": user_input})
         # 将 system_message 和用户消息输入一起发送给 LLM         llm_response = self.llm_client.get_response(messages)
     ... # 后面和确定使用哪些工具无关     
 class Tool:     """Represents a tool with its properties and formatting."""
     def __init__(         self, name: str, description: str, input_schema: dict[str, Any]     ) -> None:         self.name: str = name         self.description: str = description         self.input_schema: dict[str, Any] = input_schema
     # 把工具的名字 / 工具的用途（description）和工具所需要的参数（args_desc）转化为文本     def format_for_llm(self) -> str:         """Format tool information for LLM.
         Returns:             A formatted string describing the tool.         """         args_desc = []         if"properties" in self.input_schema:             for param_name, param_info in self.input_schema["properties"].items():                 arg_desc = (                     f"- {param_name}: {param_info.get('description', 'No description')}"                 )                 if param_name in self.input_schema.get("required", []):                     arg_desc += " (required)"                 args_desc.append(arg_desc)
         return f""" Tool: {self.name} Description: {self.description} Arguments: {chr(10).join(args_desc)} """

@classmethod def from_function(     cls,     fn: Callable,     name: str | None = None,     description: str | None = None,     context_kwarg: str | None = None, ) -> "Tool":     """Create a Tool from a function."""     func_name = name or fn.__name__ # 获取函数名
     if func_name == "<lambda>":         raise ValueError("You must provide a name for lambda functions")
     func_doc = description or fn.__doc__ or"" # 获取函数 docstring     is_async = inspect.iscoroutinefunction(fn)

async def start(self):     ... # 上面已经介绍过了，模型如何选择工具
     while True:         # 假设这里已经处理了用户消息输入.         messages.append({"role": "user", "content": user_input})
         # 获取 LLM 的输出         llm_response = self.llm_client.get_response(messages)
         # 处理 LLM 的输出（如果有 tool call 则执行对应的工具）         result = await self.process_llm_response(llm_response)
         # 如果 result 与 llm_response 不同，说明执行了 tool call （有额外信息了）         # 则将 tool call 的结果重新发送给 LLM 进行处理。         if result != llm_response:             messages.append({"role": "assistant", "content": llm_response})             messages.append({"role": "system", "content": result})
             final_response = self.llm_client.get_response(messages)             logging.info("\nFinal response: %s", final_response)             messages.append(                 {"role": "assistant", "content": final_response}             )         # 否则代表没有执行 tool call，则直接将 LLM 的输出返回给用户。         else:             messages.append({"role": "assistant", "content": llm_response})  

yum install -y python3.8update-alternatives --config python3

# 使用命令行前，请确保已经通过pip install modelscope 安装ModelScope。modelscope download --model=modelscope/ollama-linux --local_dir ./ollama-linux --revision v0.6.7

# 运行ollama安装脚本cd ollama-linuxsudo chmod 777 ./ollama-modelscope-install.sh./ollama-modelscope-install.sh

ollama serve

systemctl edit ollama.service
[Service]Environment="OLLAMA_HOST=0.0.0.0"Environment="OLLAMA_ORIGINS=*"

systemctl daemon-reloadsystemctl restart ollamasystemctl status ollamaollama run qwen3:0.6b

# 安装uvcurl -LsSf https://astral.sh/uv/install.sh | sh
# 为我们的项目创建一个新 directoryuv init mcpcd mcp
# 设置 Python 3.10+ 环境echo "3.11" > .python-version
# 创建 virtual environment 并激活它uv venvsource .venv/bin/activate
# 安装 dependenciesuv add "mcp[cli]" httpx
# 创建我们的 server filetouch web_search.py

import httpxfrom mcp.server import FastMCP
# # 初始化 FastMCP 服务器app = FastMCP('web-search')

@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': '换成你自己的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')                ifnot 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='stdio')

npx -y @modelcontextprotocol/inspector <command> <arg1> <arg2>

npx -y @modelcontextprotocol/inspector uv run web_search.py

mcp dev PYTHONFILE

mcp dev web_search.py

import asyncio
from mcp.client.stdio import stdio_clientfrom mcp import ClientSession, StdioServerParameters
# 为 stdio 连接创建服务器参数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())

import jsonimport asyncioimport osfrom typing import Optionalfrom contextlib import AsyncExitStack
from openai import OpenAIfrom dotenv import load_dotenv
from mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_client

# 初始化客户端，指向 Ollama 的本地服务client1 = OpenAI(    base_url="http://your-ip:11434/v1",  # Ollama API 地址    api_key="ollama"  # Ollama 默认无需真实 API Key，填任意值即可)#初始化客户端，指向通过api调用的Qwen3-235b-a22bclient2 = OpenAI(    # 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx",    #api_key=os.getenv("api_key"),    api_key='your-api-key',    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",)

class MCPClient:    def __init__(self):        self.session: Optional[ClientSession] = None        self.exit_stack = AsyncExitStack()        self.client1 = client1        self.client2 = client2

    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()

    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()

    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()

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())