跳到主要内容

7 篇博文 含有标签「mcp」

查看所有标签

Python 玩转 MCP:在工具函数中获取请求 Headers 的正确姿势

· 阅读需 3 分钟

一、问题出现:工具函数里怎么拿请求信息?

假设你用 FastMCP 写了一个最简单的工具函数:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My Server")

@mcp.tool(name="打招呼",description="简单的打招呼")
async def hello() -> str:
return "hello world"

这个函数很简单,运行也没问题。
但是一旦你想要在函数里获取请求头、用户信息、IP、Token 等信息(一般用于一些权限校验,数据打点等操作),就会发现:

函数没有直接的 Request 对象!

这时就要用到关键角色 —— ctx: Context

一行代码让你的 MCP 支持多机部署

· 阅读需 11 分钟

一、SSE 时代的“单机魔咒”

如果你用 Python 开发过 MCP 服务,大概率遇到过这样的场景:服务在本地跑得好好的,一到生产环境就各种幺蛾子。特别是当你想要多机部署时,发现负载均衡器怎么配置都不对劲。

问题根源:SSE 的“粘性会话”魔咒

传统的 SSE(Server-Sent Events)模式有个致命缺陷:需要维护长连接和会话状态。这意味着:

  1. 同一个客户端的请求必须始终路由到同一台服务器
  2. 负载均衡器必须配置复杂的“会话亲和性”规则
  3. 任何一台服务器宕机,正在处理的会话就彻底凉凉

为你的mcp server 添加认证功能

· 阅读需 4 分钟

使用 python 开发mcp server 非常简单,但是官方文档对于在sse 或者 streamable 服务里如何添加认证写的很不清晰,我们可以非常快速的开发一个mcp server

from mcp.server.fastmcp import FastMCP
import datetime

mcp = FastMCP("simple-mcp-server")

@mcp.tool(name="获取当前时间", description="获取当前时间")
async def get_current_time() -> str:
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")


def run():
mcp.run(transport="sse")

if __name__ == "__main__":
run()

以上是一个获取当前时间的mcp server ,很简单,只作为演示,但是如果我们不想让所有人都可以访问,需要添加一些认证信息,目前来讲,由于FastMCP 封装的太高级了,不太好添加,官方有提供基于 OAuth 的认证方式,但是有点复杂,我还没有测试成功过,本文介绍一种简单的在header 中添加认证信息。

这里分别演示 SSE 和 streamable 两种类型如何改造,原理差不太多, 通过创建一个自定义的中间件,在请求处理之前获取一下header 中的 Authorization 参数。

自己动手搭建 uv python 智能代理站

· 阅读需 13 分钟

之前有介绍在国内使用 uv 如何设置代理加速python 包的下载,在安装python 时,当时介绍了国内目前只有一个南京大学镜像站,https://mirror.nju.edu.cn/github-release/indygreg/python-build-standalone/, 这个下载站只镜像了 uv 最新的release 文件,所以对于之前月份的文件,如果使用南京大学下载站就会下载失败,本文介绍一种使用海外云服务搭建uv python 的下载代理站实现可以正常下载安装 uv python 的实现方案。

前言:为什么我们需要这个方案?

作为一名 Python 开发者,你一定遇到过这样的烦恼:想用超快的 uv 工具安装 Python,却发现国内访问 GitHub 慢如蜗牛,之前介绍过国内只有南京大学镜像站可用于加速下载,但该镜像站仅镜像了 uv 最新的 release 文件。别担心!今天我要分享的解决方案,能让你像使用国内镜像站一样流畅地安装任意版本的 uv Python!本文将详细介绍一种使用海外云服务搭建 uv Python 下载代理站的实现方案,让你轻松解决下载难题

别让你的SSE MCP服务卡住!一行 requests.post 引发的性能危机

· 阅读需 5 分钟

问题出现

在使用 Python 开发基于 MCP 的 SSE 服务时,如果不小心写了同步阻塞的代码,就非常容易踩进 IO 阻塞的大坑。来看下面这段示例代码:

from mcp.server.fastmcp import FastMCP
from pydantic import Field
import time
import requests

# 初始化mcp服务
mcp = FastMCP("hello-mcp-server")

@mcp.tool(name="url_get", description="访问某个url")
def get_url(url: str = Field(..., description="将要访问的url")):
r = requests.get(url)
return r.text

@mcp.tool(name="计算加法", description="计算加法")
def add(a: int = Field(..., description="第一个数"), b: int = Field(..., description="第二个数")):
return a + b

def run():
mcp.run(transport="sse")

if __name__ == "__main__":
run()

使用python开发mcp server之sse

· 阅读需 8 分钟

前面文章我们使用python 实现了简单的天气查询mcp server,并使用cherry studio 和 Cline 实现了mcp server 的调用,之前是使用stdio 类型的mcp,本章看一下如何开发sse类型的mcp 服务。

与stdio 的区别

sse,Server-Sent Events 是基于http 长连接的服务,以往我们在开发大模型应用的时候,由于大模型生成token 不是一次性生成的,是一个token 一个token 生成的,为了避免调用端长时间等待,会一点点的输出给调用端。

在mcp 的开发过程中,也会用到sse,但是这里的sse 和大模型输出稍微有一点不同,sse 类型的mcp 会建立一个http 监听端口,客户端在调用mcp server 时,先和mcp 建立一个长连接,之后工具的输出都是通过这个长连接发送给客户端,由于sse是单向传输,客户端只能被动的接受服务端的数据,而不能在建立连接以后再次接收客户端的数据,所以客户端在调用mcp 工具时,需要向另外一个接口发送数据。所以开发sse 的mcp 会稍微复杂一些,但是官方python sdk 中FastMCP已经做了高级封装,开发者只需要关注工具的具体实现即可。

本地stdio 类型的mcp server,通过标准的输入输出来进行数据的交互,调用端需要提前安装好环境,如python 或者 node,当然如果mcp 是一个可执行的二进制文件也可以不用安装配置环境,只是目前大部分的mcp 使用python 或者 node 进行开发的。

使用python开发mcp server

· 阅读需 12 分钟

上一篇文章介绍了UV工具的使用,有了uv 的基础,本篇我们来看一下如何使用python开发mcp server,目前主流的 mcp 分为 stdio 和 sse,stdio 为标准输入输出,通过调用工具获取工具的输出来交互,比如你在终端输入 ping host 命令,返回ping host 的输出,SSE 稍微复杂一些,通过http 接口,返回一个sse长连接,之后工具调用输入输出遵循JSONRPC规范,通过http 调用,但是结果通过SSE长连接返回,实现起来稍微复杂一些,但是调用工具和工具配置sse会更简单一些,本篇先介绍开发简单的 stdio 类型的 mcp server。工具为调用高德地图的天气查询接口返回某地的天气情况。