迟来的拥抱——向AI（2）

本文最后更新于：32 分钟前

更多参考来自MCP官方文档：https://modelcontextprotocol.io/docs/getting-started/intro

MCP SDKs: https://modelcontextprotocol.io/docs/sdk

MCP Server实现参考：https://github.com/modelcontextprotocol/server

MCP Server搭建官方文档：https://modelcontextprotocol.io/docs/develop/build-server

随便巴拉巴拉点

第一次与AI的接触是22年末疫情又一次爆发后提前回家，然后思政考试变成了线上考提交一篇论文就行，但是字数要求摆在那，让我去手搓这玩意也是让我太为难了，我于是乎就开始找一些投机的办法。当我在B站上搜索”论文”关键词的时候，一个播放量仅仅1w的视频吸引了我，那是一个和ai有关的视频，告诉了我怎么去用ai来写论文。当时的我的心情可以说是相当震惊了，不过那会还不咋会翻墙，国内媒体上关于gpt的信息也还没怎么广泛传播开来，我就到处找镜像站，体验ai

AI，多新奇呐！

写这篇笔记的时候回忆起当时似乎和女朋友在微信上提过一嘴于是去搜了下聊天记录，吼！还真找到了

让我看看我当时用的是个什么玩意，Chatsonic？这平台我在22年到26年这4年间都再也没听过了，不过确实作为了我的AI启蒙，解了我的论文的燃眉之急

再之后慢慢的也会翻墙了，用上了真正的GPT，见证了大模型的竞争、发展，从gpt到deepseek、kimi、豆包等。但是给我的感受无非就是谁更”聪明”一点，处理一些简单的任务写一点简单的代码也还不错，但是他们面对复杂的任务时就显得有点力不从心了。

直到Agent的出现，让我真真切切的感受到了AI改变世界的能力。大模型本身就应该是作为一个大脑，只不过最开始的时候我们仅仅只是给这个大脑接了个嘴巴，让他能够和用户chat，但是agent的出现给了他四肢，让大模型能够调用外部的工具来进行一些实际的操作

最开始对接四肢的方案是function calling，但是Function Calling有个比较大的缺陷，如果一个开发者写了个很好用的工具，但是他几乎不可能将他的工具直接共享给其他人，让他人直接安装到自己的平台上，原因就在于每个工具开发者自己写自己的，没有一套统一的协议去规范到底该用什么格式从大模型传递到工具调用。

MCP的出现就是为了解决这个问题，所有的工具接口都采用了一套统一的协议

MCP主要分成三部分

MCP Server

MCP Server其实可以被理解的很简单，就是一个提供特定功能（比如修图）的工具（Tools）的服务端，并暴露出接口给其他的客户端去调用内置的工具。所谓Tool，其实就是一个函数，客户端通过接口向服务端传递工具名、参数等信息，去调用Tool，随后服务端返回工具的处理结果。

这个Server与其被理解为”服务器”，不如说是”服务提供商”

体验的话可以前往vscode中安装cline插件，配置好key和api后，安装完对应的MCP Server后，模型便拥有了对应的能力

下文的内容侧重讲如何自己写一个MCP Server

首先是MCP Server也分为两大类：

按照调用方式可以将MCP Server分为本地和远程两种方式，对应的调用方式就是sdtio和sse

stdio：client直接启动脚本作为一个子进程，通过stdin向程序发送JSON格式的指令，程序通过stdout返回结果

sse：一种基于http的长连接通信方式（这会Server就真成服务端了），更适用于一个MCP server被多个客户端连接的情况，如上一篇文章提到的figma的Server。

MCP Server与MCP客户端之间的通信可以看官方文档的这一部分👉🔗👈，下面也会结合中间人代理来详细分析这一过程

下面开始结合官方文档进行一个简单的写，冲！

Claude Desktop个人版不让接入远程MCP Server，很坏，遂换成dify

第一步安装uv

1	`curl -LsSf https://astral.sh/uv/install.sh \| sh`

下面步骤就照搬官方文档吧，懒得改了，总之就是用uv创建一个新的项目

# Create a new directory for our project
uv init weather
cd weather

# Create virtual environment and activate it
uv venv
#（uv venv --python 3.12用于指定项目虚拟环境的py版本）
source .venv/bin/activate

# Install dependencies
uv add "mcp[cli]" httpx

# Create our server file
touch weather.py

将文档给出的案例直接复制到下方来，逐步来分析

from typing import Any

import httpx
from mcp.server.fastmcp import FastMCP

# Initialize FastMCP server
mcp = FastMCP("weather")

# Constants
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

async def make_nws_request(url: str) -> dict[str, Any] | None:
    """Make a request to the NWS API with proper error handling."""
    headers = {"User-Agent": USER_AGENT, "Accept": "application/geo+json"}
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return None


def format_alert(feature: dict) -> str:
    """Format an alert feature into a readable string."""
    props = feature["properties"]
    return f"""
Event: {props.get("event", "Unknown")}
Area: {props.get("areaDesc", "Unknown")}
Severity: {props.get("severity", "Unknown")}
Description: {props.get("description", "No description available")}
Instructions: {props.get("instruction", "No specific instructions provided")}
"""

@mcp.tool()
async def get_alerts(state: str) -> str:
    """Get weather alerts for a US state.

    Args:
        state: Two-letter US state code (e.g. CA, NY)
    """
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    data = await make_nws_request(url)

    if not data or "features" not in data:
        return "Unable to fetch alerts or no alerts found."

    if not data["features"]:
        return "No active alerts for this state."

    alerts = [format_alert(feature) for feature in data["features"]]
    return "\n---\n".join(alerts)


@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """Get weather forecast for a location.

    Args:
        latitude: Latitude of the location
        longitude: Longitude of the location
    """
    # First get the forecast grid endpoint
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    if not points_data:
        return "Unable to fetch forecast data for this location."

    # Get the forecast URL from the points response
    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)

    if not forecast_data:
        return "Unable to fetch detailed forecast."

    # Format the periods into a readable forecast
    periods = forecast_data["properties"]["periods"]
    forecasts = []
    for period in periods[:5]:  # Only show next 5 periods
        forecast = f"""
{period["name"]}:
Temperature: {period["temperature"]}°{period["temperatureUnit"]}
Wind: {period["windSpeed"]} {period["windDirection"]}
Forecast: {period["detailedForecast"]}
"""
        forecasts.append(forecast)

    return "\n---\n".join(forecasts)

def main():
    # Initialize and run the server
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()

FastMCP类是一个能够快速构建mcp服务器的函数，传入当前MCP服务器的名称能够返回一个mcp对象，后续注册Tools、启动Server都需要使用到这个mcp对象。

随后加入两个常量，一个是美国气象局的api地址，用于查询美国城市的天气，另一个是携带的UA

from typing import Any

import httpx
from mcp.server.fastmcp import FastMCP

# Initialize FastMCP server
mcp = FastMCP("weather")

# Constants
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

随后定义两个功能函数，用于进行一些具体的行为，通过httpx进行网络查询获取城市天气(make_nws_request)、规范化输出格式(format_alert)等等

async def make_nws_request(url: str) -> dict[str, Any] | None:
    """Make a request to the NWS API with proper error handling."""
    headers = {"User-Agent": USER_AGENT, "Accept": "application/geo+json"}
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return None


def format_alert(feature: dict) -> str:
    """Format an alert feature into a readable string."""
    props = feature["properties"]
    return f"""
Event: {props.get("event", "Unknown")}
Area: {props.get("areaDesc", "Unknown")}
Severity: {props.get("severity", "Unknown")}
Description: {props.get("description", "No description available")}
Instructions: {props.get("instruction", "No specific instructions provided")}
"""

下面开始注册工具

很容易注意到这两个新的函数上面都有@mcp.tool修饰器，该修饰器的用途便是将函数注册为工具。这个修饰器会从函数的注释里提取出这个函数的用途，以及每个参数的含义。此外还会提取函数的名字、签名等，这些信息在MCP Server与Client连接过程的初始化中就传递给了模型，让模型来判断调用这个工具的合适时机

@mcp.tool()
async def get_alerts(state: str) -> str:
    """Get weather alerts for a US state.

    Args:
        state: Two-letter US state code (e.g. CA, NY)
    """
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    data = await make_nws_request(url)

    if not data or "features" not in data:
        return "Unable to fetch alerts or no alerts found."

    if not data["features"]:
        return "No active alerts for this state."

    alerts = [format_alert(feature) for feature in data["features"]]
    return "\n---\n".join(alerts)


@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """Get weather forecast for a location.

    Args:
        latitude: Latitude of the location
        longitude: Longitude of the location
    """
    # First get the forecast grid endpoint
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    if not points_data:
        return "Unable to fetch forecast data for this location."

    # Get the forecast URL from the points response
    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)

    if not forecast_data:
        return "Unable to fetch detailed forecast."

    # Format the periods into a readable forecast
    periods = forecast_data["properties"]["periods"]
    forecasts = []
    for period in periods[:5]:  # Only show next 5 periods
        forecast = f"""
{period["name"]}:
Temperature: {period["temperature"]}°{period["temperatureUnit"]}
Wind: {period["windSpeed"]} {period["windDirection"]}
Forecast: {period["detailedForecast"]}
"""
        forecasts.append(forecast)

    return "\n---\n".join(forecasts)

最后这部分代码加上，让程序能够启动时运行MCP Server

需要注意的点在mcp.run后的参数transport中，这个参数的值为stdio，说明mcp Server是以前文提到的标准输入输出的方式启动的

def main():
    # Initialize and run the server
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()

我们使用MCP inspector来调试一下这个程序

终端输入

1	`npx @modelcontextprotocol/inspector`

很简单便可以启动一个MCP inspector的服务，浏览器访问携带AUTH_TOKEN的url6274端口

由于当前我的MCP Server是以stdio的方式本地启动的，因此就使用uv run server.py的方式来跑

随后便来到了这样一个界面，下方的History里可以看到每一次操作的请求和响应，有利于我们来对工具调用的流程去进行一个分析

我们来到Tools

点击List Tools后，我们能看到下方列举出了我们定义的两个工具函数。与此同时History中出现了新的一条记录，我们展开来分析

Run Tool用于手动传参运行Tool

不过我们这里对接用的应用是dify，而dify仅支持sse方式来添加MCP Server

我们上面的程序改为sse相当的简单，进需要将transport=’stdio’改为transport=’sse’即可，运行uv run weather.py便可以启动http服务

如果有改动端口和host的需求的话应当在FastMCP初始化时修改，而非在run()方法中

1	`mcp = FastMCP("weather", port=8001, host='0.0.0.0')`

SSE架构的Server启动后的URL有下面两个：

连接地址： http://127.0.0.1:8000/sse
消息地址：*http://127.0.0.1:8000/messages

MCP inspector或是dify里添加MCP Server的时候都需带上/sse路径来连接

成功添加后dify的右侧会出现工具列表

ai最懂ai，需求讲给隔壁gemini后让他生成一段简短的提示词（当然dify里也内置了ai生成提示词的功能）

将MCP Server中的工具添加进来后，配置好模型，试着体验了一手，嗯！跑起来了！

为了更加直观看到一整个传输过程，我让gemini帮忙写了个代理，用来打印出MCP Server和dify之间的通信过程

该代理监听8000端口并代理8001端口的sse，因此我们添加mcp server url时应当填入的是8000端口

import uvicorn
import httpx
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse

app = FastAPI()

TARGET_URL = "http://127.0.0.1:8001"


@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"])
async def proxy(request: Request, path: str):
    url = f"{TARGET_URL}/{path}"

    # --- 1. 捕获并打印请求体 ---
    method = request.method
    headers = dict(request.headers)
    headers.pop("host", None)
    params = request.query_params

    # 读取请求体内容
    body_bytes = await request.body()
    body_str = body_bytes.decode('utf-8', errors='ignore')

    print(f"\n{'#' * 30} 收到客户端请求 {'#' * 30}")
    print(f"URL    : {method} {url}")
    print(f"Params : {params}")
    print(f"Headers: {headers}")
    print(f"Body   : {body_str if body_str else '[Empty Body]'}")
    print(f"{'#' * 68}\n")

    # --- 2. 转发请求 ---
    client = httpx.AsyncClient(timeout=None)
    req = client.build_request(
        method=method,
        url=url,
        headers=headers,
        params=params,
        content=body_bytes  # 转发原始字节流
    )

    try:
        response = await client.send(req, stream=True)

        print(f">>> 目标服务响应状态码: {response.status_code}")
        print(f">>> 开始透传响应流...")

        async def event_generator():
            try:
                async for chunk in response.aiter_bytes():
                    if chunk:
                        # 打印响应流片段
                        print(chunk.decode('utf-8', errors='ignore'), end="", flush=True)
                        yield chunk
            except Exception as e:
                print(f"\n[流式传输中断]: {e}")
            finally:
                await response.aclose()
                await client.aclose()
                print(f"\n{'=' * 20} 传输会话结束 {'=' * 20}")

        return StreamingResponse(
            event_generator(),
            status_code=response.status_code,
            headers=dict(response.headers)
        )

    except Exception as e:
        await client.aclose()
        print(f"转发失败: {e}")
        return StreamingResponse(iter([f"Proxy Error: {str(e)}".encode()]), status_code=502)


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000, timeout_keep_alive=60)

当我使用这个代理url作为MCP Server，向Agent发起会话查询纽约市的天气预报时，首先收到一个空请求体的GET请求，在SSE中该请求用于获取后续的消息会话地址，返回一个带有session id的path

随后客户端（dify），开始通过POST请求开始初始化了，初始化过程中等于客户端和服务端打了个招呼，clientInfo字段中携带了客户端的名称以及版本号，告诉服务端诶咱们这个请求来自Dify的1.13.0版本。用的mcp版本是2025-06-18更新的最新版

既然收到了客户端的请求，那么我们的服务端自然是要做出响应的。匹配上后发现MCP版本号是一致的！那么后续的沟通就畅通无阻，prompts和resources我们暂时还没配置到，因此都是false告诉客户端我们服务端这些功能是不支持的。同时还告知了客户端Server的名字为weather以及版本号

随后客户端又发起了一个请求，告知服务端：”收到！”

紧接着客户端又发起一个请求，这个请求不同于之前的打招呼。前面打完招呼后，客户端就要开始干活了，于是开始向服务端申请工具调用的能力了，调用tool是get_forcast，并且传递经纬度参数来查询天气预报。这时候细心的朋友肯定会问了：”诶！这客户端咋知道服务端有什么工具呢？为啥就知道get_forcast这个工具的调用方法呢？”，答案就藏在加载MCP Server URL的一瞬间，这里就不做展示了，客户端在那时也向服务端打了招呼，并获取到了服务端的工具列表，不过后面为了更快，减少不必要的步骤，客户端就将工具列表保存在了本地缓存中。所以在真正调用工具时客户端并没有再去问一遍服务端”诶你有什么工具呀”，而是大模型直接利用本地缓存的工具列表直接用了

随后服务端向客户端返回了响应信息，包含了格式化后的详细天气预报情况

一次完整的交互到此就结束了，随后dify就会将这个最终的响应发给大模型，让模型去做最终的总结

提一嘴如果是使用stdio启用的mcp server，还可以在终端中直接向mcp server发送json来做沟通，效果是一样的

也许有些人会有这么一种感觉：讲了这么多但是似乎和大模型都没什么关系？

在这篇文章的主要角色MCP下，这个答案是”对”。MCP这个协议实际上并没有去规范模型到底应该如何去调用工具，它只是负责规范MCP Host与Server之间通信格式的一种协议，但是大模型和Host之间的通信，每一种MCP Host都有自己不同的实现方案，与MCP无关，这便来到了agent的内容了

MCP Host/Client

MCP Server段中的cline实际上就是一个agent。vscode就可视为一个MCP Host（主机）

vscode充当Host，当和Server建立连接时，vscode运行时会实例化一个MCP Client对象来维护连接。

现在很多时候Host和Client之间的界限也没分的那么清了，常常放在一块说。关于这二者的开发这里不做赘述了就

迟来的拥抱——向AI（3）上一篇

迟来的拥抱——向AI（1）下一篇