本文介绍了MCP（模型上下文协议）这一新兴的开放协议，它作为AI应用程序与外部世界交互的桥梁，简化了开发流程并降低了集成复杂性。文章通过手把手教学，引导读者使用Trae等工具快速体验MCP Server，并进一步指导开发者搭建本地Client和Server，包括安装工具和环境依赖、创建MCP Client项目、安装MCP SDK以及创建并运行client.py和rag_server.py文件，最终实现简易的RAG功能调用。

🔑**MCP协议简介:** MCP（模型上下文协议）是一种新兴的开放协议，旨在作为AI应用程序与外部世界交互的桥梁，类似于AI领域的“USB-C接口”，实现AI模型与各种外部数据源和工具的无缝集成，简化开发流程，降低集成复杂性。

🛠️**快速体验MCP Server:** 可以借助Trae，cursor等工具快速体验mcp server，文章以Trae作为演示，通过选择合适的MCP Server（如高德的mcp server），配置API Key，并新建智能体，即可实现如长沙三日游旅游攻略的制定等功能。

💻**本地Client和Server搭建:** 开发者可以本地部署Client，首先需要安装uv等工具及环境依赖，然后创建MCP Client项目和虚拟环境，安装MCP SDK，并创建client.py文件，用于连接到多个MCP服务器并处理查询。同时，创建本地server，如rag_server.py，实现简单的RAG功能调用，并通过运行相应的命令启动client.py和rag_server.py文件。

引言：

在 AI 技术飞速发展的当下，MCP（Model Context Protocol，模型上下文协议）作为一种新兴的开放协议，正逐渐成为 AI 应用程序与外部世界交互的重要桥梁。它就像 AI 领域的“USB-C 接口”，使 AI 模型能够与各种外部数据源和工具实现无缝集成，极大地简化了开发流程，降低了集成复杂性。对于开发者而言，掌握 MCP 的基本原理并能够快速编写出简易的 Client 和 Server，是迈向高效 AI 开发的重要一步。本文将带你走进 MCP 的世界，手把手教你搭建起一个简单的 Client-Server 系统，让你轻松上手体验 AI 应用开发的魅力

快速上手MCP

我们可以借助Trae，cursor等快速体验mcp server，此处以Trae作为演示。

1.获取mcp server

以下是几个mcp server市场：
Awesome MCP Server
Smithery
MCP.so
我们用mcp.so进行演示，点开网站，可以通过具体需求去选择自己想要的servers，此处我们使用高德的mcp server实现长沙三日游的旅游攻略的制定

点开后，我们将框中的代码进行复制,注意将api-key改为自己的
获取key

2.配置mcp server

在trae选择builder with mcp后点击添加点击手动配置后，将前面复制的代码粘贴进去

添加成功后，如下图所示

在进行旅游规划时，需要新建一个智能体（默认的只能完成编程类任务），以下是运行效果：

打造本地client和server

以上的配置需要依靠cursor，trae等进行配置，那么有没有一种方法可以本地进行client的部署呢（有的，兄弟有的【手动狗头】）

1.安装工具及环境依赖

1.1 uv安装流程

# 采用pip命令进行安装pip install uv

2.简易mcp-client创建

2.1 创建mcp-client项目

# 创建目录uv init mcp-clientcd mcp-client

2.2 创建虚拟环境

# 创建并启用虚拟环境uv venv venv_name # 创建虚拟环境# 激活虚拟环境source venv_name/bin/activate # Linux/macosvenv_name/Scripts/activate # Windows

安装mcp sdk

uv add mcp

2.3 创建client.py文件

import sysimport ollamaimport asyncioimport osimport jsonfrom typing import Optional, Dict, List, Tuplefrom contextlib import AsyncExitStackfrom openai import OpenAIfrom dotenv import load_dotenvfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientload_dotenv()class MCPClient:    def __init__(self):        """初始化 MCP 客户端"""        self.exit_stack = AsyncExitStack()        self.openai_api_key = your_api_key        self.base_url = your_url        self.model = your_model_name        if not self.openai_api_key:            raise ValueError("api_key 未设置")        self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)        # 存储多个服务器会话        self.sessions: Dict[str, ClientSession] = {}        # 存储工具名称到服务器会话的映射，用于知道调用哪个服务器        self.tool_to_session: Dict[str, ClientSession] = {}        self.exit_stack = AsyncExitStack()    async def connect_to_server(self, server_script_path: str, server_id: str = None):        """连接到 MCP 服务器并列出可用工具                Args:            server_script_path: 服务器脚本路径            server_id: 服务器标识符，如果为None则使用脚本路径作为标识符        """        if server_id is None:            server_id = server_script_path                    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))        stdio, write = stdio_transport        session = await self.exit_stack.enter_async_context(ClientSession(stdio, write))        await session.initialize()                # 保存会话到字典中        self.sessions[server_id] = session        # 列出 MCP 服务器上的工具        response = await session.list_tools()        tools = response.tools                # 将每个工具映射到对应的服务器会话        for tool in tools:            self.tool_to_session[tool.name] = session                    print(f"\n已连接到服务器 {server_id}，支持以下工具:\n", [tool.name + '\n' for tool in tools])                return tools    async def connect_to_multiple_servers(self, server_paths: List[str]):        """连接到多个MCP服务器                Args:            server_paths: 服务器脚本路径列表        """        all_tools = []        for i, path in enumerate(server_paths):            server_id = f"server_{i+1}"            tools = await self.connect_to_server(path, server_id)            all_tools.extend(tools)                    print(f"\n总共连接了 {len(server_paths)} 个服务器，共支持 {len(all_tools)} 个工具")        return all_tools    async def process_query(self, query: str) -> str:        messages = [{"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": query}]        try:            last_messages = f'用户需求：{query}\n以下是补充内容：\n'                        # 收集所有服务器的工具            available_tools = []            for session_id, session in self.sessions.items():                response = await session.list_tools()                for tool in response.tools:                    available_tools.append({                        "type": "function",                         "function": {                            "name": tool.name,                            "description": tool.description,                            "input_schema": tool.inputSchema                        }                    })                        print(f"可用工具总数: {len(available_tools)}")            while True:                response = self.client.chat.completions.create(                    model=self.model,                    messages=messages,                    tools=available_tools                )                # response = ollama.chat(                #     model='EntropyYue/chatglm3:latest',                #     messages=messages,                #     tools=available_tools                # )                # 处理返回的内容                content = response.choices[0]                print(content)                                # 如果不需要调用工具,直接返回结果                if content.finish_reason != "tool_calls":                    messages = [{"role": "user", "content": last_messages}]                    response = self.client.chat.completions.create(                        model=self.model,                        messages=messages                    )                    # response = ollama.chat(                    #     model='EntropyYue/chatglm3:latest',                    #     messages=messages                    # )                    return response.choices[0].message.content                                    # 需要调用工具,则继续处理                messages.append(content.message.model_dump())                                # 处理所有工具调用                for tool_call in content.message.tool_calls:                    tool_name = tool_call.function.name                    tool_args = json.loads(tool_call.function.arguments)                    # 根据工具名查找对应的服务器会话                    if tool_name in self.tool_to_session:                        session = self.tool_to_session[tool_name]                        # 执行工具                        result = await session.call_tool(tool_name, tool_args)                        print(f"\n\n[Calling tool {tool_name} with args {tool_args}]\n\n")                        print(result)                        print("#########################")                        messages.append({                            "role": "tool",                            "content": result.content[0].text,                            "tool_call_id": tool_call.id,                        })                        last_messages += result.content[0].text                    else:                        error_msg = f"找不到工具 {tool_name} 对应的服务器"                        print(f"\n\n[错误] {error_msg}\n\n")                        messages.append({                            "role": "tool",                            "content": error_msg,                            "tool_call_id": tool_call.id,                        })                        last_messages += error_msg                # 循环继续,让模型处理工具调用的结果        except Exception as e:            return f"⚠️ 调用 OpenAI API 时出错: {str(e)}"    async def chat_loop(self):        """运行交互式聊天循环"""        print("\nMCP 客户端已启动！输入 'quit' 退出")        while True:            try:                query = input("\nQuery: ").strip()                if query.lower() == 'quit':                    break                response = await self.process_query(query)                print(f"\n🤖 OpenAI: {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 <server_script_path1> [<server_script_path2> ...]")        sys.exit(1)            client = MCPClient()    try:        # 连接到所有提供的服务器        server_paths = sys.argv[1:]        await client.connect_to_multiple_servers(server_paths)        await client.chat_loop()    finally:        await client.cleanup()if __name__ == "__main__":    asyncio.run(main()) 

运行client.py文件

uv run client.py {your_server}

2.4 创建本地server

此处我们采用简单的rag功能调用，作为示例，以下是rag_server.py

from mcp.server.fastmcp import FastMCPfrom langchain_huggingface import HuggingFaceEmbeddingsfrom langchain_community.vectorstores import FAISSimport asyncio# 初始化 MCP 服务器mcp = FastMCP("RAGServer")EMBEDDING_MODEL = your_embedding_model # 替换为自己的embedding_modelembeddings = HuggingFaceEmbeddings(model_name=EMBEDDING_MODEL)db = FAISS.load_local('your_faiss',                      embeddings, allow_dangerous_deserialization=True)async def format_related_content(related_docs):    """格式化相关文档内容"""    return "\n".join([doc.page_content.replace("\n\n", "\n") for doc in related_docs])async def fetch_insurance_knowledge(query: str):    """从 FAISS 数据库中检索相关知识"""    try:        # return "进入工具函数"        docs = db.similarity_search(query, k=10)        docs = await format_related_content(docs)        return docs    except Exception as e:        return f"查询失败：{str(e)}"@mcp.tool()async def get_insurance_knowledge(query: str):    """    输入你需要查询的问题，返回相关答案。    :param query: 查询语句（需使用中文）    :return: 格式化后的相关答案    """    try:        # return "进入工具"        data = await fetch_insurance_knowledge(query)        return data    except Exception as e:        return f"查询失败：{str(e)}"if __name__ == "__main__":    # 以标准 I/O 方式运行 MCP 服务器    asyncio.run(mcp.run(transport="stdio"))

运行命令

uv run client.py rag_server.py