一文读懂 MCP 协议与求职助手案例实战

补充 Tools 细节

通过上篇文章，我们知道：AI 模型在垂直领域和及时更新的知识，依赖于外部工具执行后的结果（比如，明天北京天气，浪浪山中的动画人物形象）。

并且把执行结果追加到上下文中，大模型才会基于相关内容做进一步的动作。

为了方便对比，把上一篇案例中的tools.py 部分代码单独放一下——

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49


import os
from tavily import TavilyClient

tavily_client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))

tools = [
    {
        "name": "tavily_search",
        "description": "使用 Tavily 进行网络搜索，获取相关信息",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "搜索查询字符串",
                }
            },
            "required": ["query"]
        },
    },
]


def tavily_search(query: str) -> str:
    """
    使用 Tavily 进行网络搜索并返回搜索结果。
    参数:
        query (str): 搜索查询字符串
    返回:
        str: 搜索结果的字符串表示
    """
    try:
        response = tavily_client.search(query, search_depth="advanced", max_results=5)
        results = response.get("results", [])
        
        if not results:
            return "未找到相关搜索结果。"
        
        # 格式化搜索结果
        formatted_results = []
        for result in results:
            title = result.get("title", "无标题")
            url = result.get("url", "")
            content = result.get("content", "")
            formatted_results.append(f"标题: {title}\n链接: {url}\n内容: {content}\n")
        
        return "\n".join(formatted_results)
    except Exception as e:
        return f"搜索过程中发生错误: {str(e)}"

“tools” 变量是个对象列表，用来填充提示词中{tools} 部分

为什么需要这部分？

原因也不复杂：在大模型不支持 function calling之前，只能通过这种提示词注入的方式告诉模型：有什么工具可以调用以及如何调用。

而当大模型支持 function calling 后，这个步骤没有省略——而是从提示词中迁移到大模型提供的**专门接口“tools”**中。

长这样——

结构上看，只是比上文的tools 多了一层封装——{“type”: “function”, “function”: {原工具提示词部分内容} }

回到 “def tavily_search(query: str)” 这部分

不难看出：所谓的"工具"本质上也是一个函数。

那LLM是怎么“调用”tools执行具体任务的呢？

在大模型不支持 function calling之前：就以LLM 输出的 “Action：XXX” 内容为条件，做if-else 选择。

简化后逻辑是这样——

1
2
3


if ( LLM 输出 "Action: tavily_search" ):

    tavily_search( query = "Action Input: 工具参数" )

当大模型支持 function calling后，大模型则直接在"tool_calls“中返回它希望调用的具体工具（函数）名和相关参数。长这样——

工具执行阶段：只需调用"name"指定的函数，按照 “arguments” 中的参数执行——

1
2
3
4
5
6
7


for tool_call in response.tool_calls:
  tool_name = tool_call["name"]
  tool_args = tool_call["args"]
  get_tool = tools_with_name[tool_name]
  print(f"调用工具：{tool_name}, {tool_args}")
  # 执行工具函数（同步）
  call_tool_ret = get_tool.invoke(tool_args)

好，以上补充了 tools 在Agent中执行的细节——重点是大模型根据任务自主决策使用什么工具，然后由外部工具执行具体任务。

工具执行后的内容，又会作为“记忆”保存在上下文中，方便下一轮对话中大模型进一步分析、解答用户提问。

不难看出按照上面的框架，可以在一个Agent 中扩展多个工具——因为，只要继续追加 tools 就好了嘛。

MCP 出世

如果有多个的Agent 都需要调用 tavily_search 这个工具，会怎样？

没错！需要在每个Agent的tools.py代码/workflow 中都添加这个tavily_search 工具。

如果公司里就我一个 Agent 开发人员，可能问题不大。

但是，如果公司有10几个开发人员负责开发10个不同的Agent，一个小小的变更带来的沟通、修改、维护成本都会很高。

这样，MCP的需求场景就清晰了（让Agent和 Tools 间实现松耦合的结构）：

同一个工具函数，需要在不同地方重复调用
同一个工具函数，分散在各个项目中，修改和管理分散
如果可以将这些工具集中管理，一方面可以让使用者和执行者解耦合（分离）
另一方面，分离后还可以进一步做使用人员的权限控制，比如：这个工具开发人员A 可以用，B用不了。

而分离后，LLM 与 Tools 间通信的协议，就是我们今天的主角 —— MCP。

没错，MCP 本质上不是工具(Tools) 也不是大模型新增的功能模块，而是Agent与tools 函数间的“桥梁”——通信协议。

该协议本着有什么用什么的原则：

在本地，使用STDIO 实现两个进程间通信；跨设备，使用HTTP 协议实现两个应用间的通信。

前者，还是只能本地执行工具，无法把“分离”的优势发挥到最大。

后者因为发展原因，又用到的了两个子协议：Server-Sent Events (简称：SSE) 和 Streamable。

它们都使用 HTTP 协议，但 SSE 是基于 HTTP 的文本消息推送协议，而 Streamable 是基于 HTTP 的二进制内容传输技术。就像电子邮件（SSE）和网盘下载（Streamable）都使用 HTTP，但使用场景还是有差异的。

从Agent的视角看

工具函数的执行方从本地转移到了MCP server 上，详见下图 3.1-3.4 部分：

MCP server 中执行工具的方法有多种：

可以在mcp server 上执行本地函数

也可以在mcp server 上调用第三方服务的API接口

也可以在mcp server 上请求大模型

这里MCP协议的作用有两个

在Agent 初始化阶段，MCP server 提供工具和相关的描述信息——称为 “tools/list”

例如：

下图记录了MCP client 与MCP server 间成功建立连接后，通过 “ListTools” 获取到的四个工具（get_joblist_by_expect_job, get_job_by_resume, get_word_by_filepath, fix_resume）和每个工具的描述信息：
在Agent 接收LLM返回的工具调用信息后，由MCP Client 发起工具调用——称为 “tools/call”

从图中可以看到：MCP Client向MCP Server的工具"get_joblist_by_expect"发送如下参数——
1

{"job": "AI应用工程师"}

MCP server 的实现

好了，MCP出现的背景和工作流程我们讲清楚了。下面重点放在MCP server 的实现上。

这里以一个“求职助手” MCP Server为例：从 0-1 实现这个功能。

该助手可以帮助求职者在职位筛选和简历修改上，使用大模型分析、识别与求职者匹配度高的岗位，并输出优化后的个人简历。

该MCP Server 主要提供以下四个工具：

工具名	工具描述	工具参数	调用LLM
get_joblist_by_expect_job	根据求职者的期望岗位获取岗位列表数据	job: [职位名]	N
get_job_by_resume	根据岗位列表以及求职者的简历获取适合求职者的三个岗位及提供相关求职建议	jobs: [职位清单], resume: [个人简历]	Y
get_word_by_filepath	读取指定路径的word文件	filepath: [word 简历路径]	N
fix_resume	根据目标岗位要求改写并完善个人简历	jd: [目标职位jd], resume: [个人简历]	Y

目标是帮助求职者筛选出匹配度高的职位并输出优化后的简历，关系图如下——

0. LLM Client & 提示词准备

如上文所说 “get_job_by_resume” 和 “fix_resume” 这两个工具是需要调用 LLM完成简历内容处理的，所以需要准备一个LLM 调用模块—— “LLMClient”。

模块中，提前准备了 deepseek-chat 模型的接口和 API 参数。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29


import os
import logging
from openai import OpenAI
from dotenv import load_dotenv

class LLMClient:
    def __init__(self, logger: logging.Logger):
        self.logger = logger
        self.client = self._get_client()

    def _get_client(self)->OpenAI:
        load_dotenv()
				# 根据自己环境，可以选择不同的模型提供方
        client = OpenAI(
            api_key=os.getenv("DEEPSEEK_API_KEY"),
            base_url="https://api.deepseek.com",
            # api_key=os.environ.get("GEMINI_API_KEY"),
            # base_url="http://ollama.host:8045/v1"
            )
        return client

    def send_messages(self, messages):
        # 根据不同的模型提供方，选择具体的模型
        response = self.client.chat.completions.create(
            model="deepseek-chat",
            # model="gemini-3-flash",
            messages=messages,
        )
        return response

和之前ReACT 案例一样，这里的提示词模板也使用了占位符——{resume} {job_list} {input}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35


Job_Search_Prompt = """
【AI求职助手】
你是一个AI求职助手, 我正在寻找与我的技能和经验相匹配的工作机会。以下是我的简历摘要和搜集到的岗位需求列表

【个人简历】
{resume}

【岗位需求列表】
{job_list}

请帮我匹配最合适的3个岗位, 并根据我的简历提供简要的求职建议。
"""

ResumePrompt = """
你是一个 AI 简历助手。我会给你提供我的简历以及某公司的详细岗位要求。你的任务是根据公司的岗位要求, 帮我改写和完善我的简历，使我的简历符合该公司的要求。
此外，我还会给你一个简历模板，模板中会包含简历中部分内容的大纲，当你匹配到我的简历中有模板提及的内容时，要按照我模板的格式进行编写。

简历：
{resume}

简历模板：
专业技能
  请在此描述符合职位要求的技能，尤其是AI方向的技能

项目经验
 (1) 项目描述
 (2) 我在项目中的角色
 (3) 项目规模
 (4) 技术堆栈
 (5) 已开发模块的描述
 (6) 解决难题的经验

岗位要求：
{input}
"""

1. 本地工具

这里以后两个工具（get_word_by_filepath、fix_resume）为例——

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24


class JobTools(LLMClient):
    def register_tools(self, mcp: Any):
        """Register job tools."""
            
        @mcp.tool(description="读取指定路径的word文件")
        def get_word_by_filepath(filepath: str) -> list:
            """根据文件路径获取word文件内容"""
            content=read_word_file(filepath)
            return content

        @mcp.tool(description="根据目标岗位要求改写并完善个人简历")
        def fix_resume(jd: str, resume: str) -> str:
            """根据目标岗位要求改写并完善个人简历"""
            #将待优化简历以及目标岗位jd信息注入到 prompt 模板
            prompt = ResumePrompt.format(resume=resume,input=jd)
            messages = [{"role": "user", "content": prompt}]
            
            self.logger.info(f"prompt: {prompt}")

            #发送给 LLM
            response = LLMClient.send_messages(self,messages)
            response_text = response.choices[0].message.content

            return response_text

不难看出：MCP server 中的工具和开头举例的 “def tavily_search(query: str) " 工具一样——还是函数。

只是，它被装饰器 (@mcp.tool) 装饰过后，它多了一些mcp tool所需的元数据（工具名、工具描述、工具参数）

mcp_tools_list

另外，fix_resume 这个工具在执行过程中需要调用 LLM，所以工具中进行了模型消息拼装——

1
2
3
4
5


prompt = ResumePrompt.format(resume=resume,input=jd)
messages = [{"role": "user", "content": prompt}]

#拼装好后 发给上一步中 准备好的LLMClient
response = LLMClient.send_messages(self,messages)

2. MCP Server 封装实现

最后，使用mcp中的FastMCP 帮助我们快速实现 MCP server 的组装。

从外到里看包含：http入口、http安全配置、认证中间件、server sse初始化、工具注册。

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106


import os
import logging
import argparse
from mcp.server.fastmcp import FastMCP
from mcp.server.transport_security import TransportSecuritySettings
from .tools.job import JobTools
import uvicorn
from starlette.responses import JSONResponse


class JobSearchMCPServer:
    def __init__(self):
        self.name = "jobsearch_mcp_server"

        # 配置传输安全设置，允许本地主机和自定义域名
        transport_security = TransportSecuritySettings(
            enable_dns_rebinding_protection=False,  # 禁用 DNS rebinding protection 以支持动态域名
            allowed_hosts=[
                "127.0.0.1",
                "localhost",
                # 本机局域网ip
                "192.168.*.*"
            ]
        )

        # 初始化 FastMCP
        self.mcp = FastMCP(
            "StatelessServer",
            stateless_http=False,
            transport_security=transport_security
        )

        # 开启日志
        logging.basicConfig(
            level=logging.INFO,
            format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        )
        self.logger = logging.getLogger(self.name)

        # 调用工具注册函数
        self._register_tools()

    def _register_tools(self):
        """Register all MCP tools."""
        job_tools = JobTools(self.logger)
        job_tools.register_tools(self.mcp)

    def get_app(self, host: str, port: int):
        """Get the ASGI application for SSE-based MCP server."""
        self.logger.info(
            f"Starting MCP Server in SSE-based mode on {host}:{port}"
        )
        self.logger.info(f"MCP Server is accessible at http://{host}:{port}")
        # 使用 sse 模式对外提供 MCP 服务
        app = self.mcp.sse_app()
        
        # 使用 ASGI 包装器直接实现认证，避免 BaseHTTPMiddleware 处理流式响应的问题
        async def auth_wrapper(scope, receive, send):
            mcp_token = os.getenv("MCP_AUTH_TOKEN")

            if scope["type"] == "http":
                path = scope.get('path', '')
                method = scope.get('method', '')
                
                # 添加详细日志，观察到底是哪个路径在报错
                self.logger.info(f"Incoming request: {method} {path}")

                # 获取请求头
                auth_header = None
                for key, value in scope.get("headers", []):
                    if key.lower() == b"authorization":
                        auth_header = value.decode("utf-8")
                        break
                
                # 验证逻辑
                if not auth_header or not auth_header.startswith("Bearer "):
                    self.logger.warning(f"Unauthorized: {method} {path}")
                    response = JSONResponse({"error": "Unauthorized"}, status_code=401)
                    await response(scope, receive, send)
                    return
                
                # 校验 Token
                try:
                    token = auth_header.split(" ")[1]
                    if token != str(mcp_token):
                        raise ValueError("Token mismatch")
                except Exception:
                    response = JSONResponse({"error": "Invalid Token"}, status_code=401)
                    await response(scope, receive, send)
                    return
            
            # 认证通过，交给 FastMCP 的 SSE App 处理
            await app(scope, receive, send)
        # 最终，返回 ASGI 应用
        return auth_wrapper


def main():
    parser = argparse.ArgumentParser(description="Run MCP SSE-based server")
    parser.add_argument("--host", default="0.0.0.0", help="Host to bind to")
    parser.add_argument("--port", type=int, default=8000, help="Port to listen on")
    args = parser.parse_args()

    server = JobSearchMCPServer()
    app = server.get_app(host=args.host, port=args.port)
    uvicorn.run(app, host=args.host, port=args.port)

身份认证 token 写在环境变量MCP_AUTH_TOKEN 中——

localenv

让AI 画个逻辑图，更容易理解一些——

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


请求进来 (任何 path)
    ↓
auth_wrapper (ASGI 中间件)
    ├─ 没带 Authorization → 401 {"error": "Unauthorized"}
    ├─ Token 不匹配 → 401 {"error": "Invalid Token"}
    └─ Token 正确 → 放行
            ↓
    FastMCP.sse_app() (MCP 协议处理)
            ↓
    JobTools 注册的所有工具

至此整个 MCP server 就搭建完了。

最终项目结构长这样——

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17


jobsearch_mcp_server/
├── .env
├── pyproject.toml
├── src/
│   └── jobsearch_mcp_server/
│       ├── job.txt
│       ├── __init__.py
│       ├── server.py          # 主服务
│       ├── llm/
│       │   └── llm.py
│       ├── prompt/
│       │   └── prompt.py
│       ├── tools/
│       │   └── job.py
│       └── word/
│           ├── word.py
│           └── 个人简历.docx

使用 uv 工具启动项目——

1
2
3
4


# run mcp server
pip install uv
uv sync
uv --directory S:\trae_pj\jobsearch_mcp_server-1.0.0\src\jobsearch_mcp_server run jobsearch-mcp-server

调用效果

因为本文重点在MCP server 的实现，Agent部分的差异和设计不表。

测试时还是请出老朋友 chatbox。

支持 MCP Client的工具都可以。例如：Trace、Cursor、Roo、Claude Code。。。

1. MCP Client 配置

Client 中的配置内容主要就是： mcp server 的http 地址、接口和认证凭据。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12


{"mcpServers":
 {
   "remote_jobsearch":
   {
     "url":"http://192.168.202.100:8000/sse",
     "headers":
     {
       "Authorization":"Bearer key_567890"
     }
   },
 }
}