当前位置： 首页 > news >正文

知识点3：python-sdk 核心概念（prompt、image、context）

news 2025/7/18 6:38:21

0 前言

官方网址：https://github.com/modelcontextprotocol/python-sdk
https://modelcontextprotocol.io/introduction
所有内容均以官方为主，可结合食用～
系统：ubuntu 20.04

1 Prompts

提示是可重复使用的模板，可帮助 LLM 有效地与您的服务器交互：

from mcp.server.fastmcp import FastMCP  # 导入 MCP 服务端框架主类
from mcp.server.fastmcp.prompts import base  # 导入基础消息类型（用于多轮对话）mcp = FastMCP(name="Prompt Example")  # 创建一个 MCP 服务实例，名称为 "Prompt Example"# 注册一个 prompt（提示），用于代码审查
@mcp.prompt(title="Code Review")  # 指定 prompt 的标题为 "Code Review"
def review_code(code: str) -> str:# 输入参数 code 为待审查的代码return f"Please review this code:\n\n{code}"  # 返回格式化的代码审查请求文本# 注册一个 prompt（提示），用于调试助手
@mcp.prompt(title="Debug Assistant")  # 指定 prompt 的标题为 "Debug Assistant"
def debug_error(error: str) -> list[base.Message]:# 输入参数 error 为用户遇到的错误信息return [base.UserMessage("I'm seeing this error:"),  # 用户消息：说明遇到错误base.UserMessage(error),                     # 用户消息：具体错误内容base.AssistantMessage("I'll help debug that. What have you tried so far?"),  # 助手消息：询问用户已尝试的解决方法]# 返回一个消息列表，模拟多轮对话场景，便于后续 AI 或助手进if __name__ == '__main__':mcp.run(transport='sse') # transport 用来指定通讯协议# 用于启动 MCP（Model Context Protocol）服务，并指定通信方式为 SSE（Server-Sent Events）。

run反法查看原码：有三个可选参数["stdio", "sse", "streamable-http"]，默认为"stdio"

def run(self,transport: Literal["stdio", "sse", "streamable-http"] = "stdio",mount_path: str | None = None,) -> None:

补充：prompt装饰器参数可以更完整@mcp.prompt(name="debug", title="Debug Assistant", description="帮助调试错误")
在这里插入图片描述

2 transport

MCP 使用 JSON-RPC 编码消息。JSON-RPC 消息必须采用 UTF-8 编码。
该协议目前为客户端-服务器通信定义了两种标准传输机制：

  • stdio，通过标准输入和标准输出进行通信
  • 可流式传输的 HTTP（sse就包含在这部分）

客户端应该尽可能支持 stdio，客户端和服务器也可以以可插拔的方式实现自定义传输。

2.1 stdio(标准输入输出)

在stdio传输中：

  • 客户端将 MCP 服务器作为子进程启动。(python中的表现为直接启用客户端即可，通过params配置服务器子进程。)
  • 通信通道建立：服务器从其标准输入（stdin）读取 JSON-RPC 消息，并将消息发送到其标准输出（stdout）。
    • 客户端 -> 服务器： 客户端通过写入服务器的 stdin 管道来发送消息。服务器通过读取自己的 stdin 来接收这些消息。
    • 服务器 -> 客户端： 服务器通过写入自己的 stdout 管道来发送消息。客户端通过读取服务器的 stdout 来接收这些消息。
  • 消息格式：消息是单独的 JSON-RPC 请求、通知或响应。
    • 请求 (Request)： 客户端调用服务器上的一个方法，并期望一个响应
    • 通知 (Notification)： 客户端调用服务器上的一个方法，但不期望任何响应
    • 响应 (Response)： 服务器对客户端之前发送的请求的答复
  • 消息分隔：由于管道是字节流，需要一种方法来区分一个消息的结束和下一个消息的开始。这里采用的分隔符是换行符 (\n)。
    • 发送方（无论是客户端通过 stdin 还是服务器通过 stdout）在写完一个完整的 JSON-RPC 消息后，必须紧接着写入一个换行符 (\n)。
    • 接收方（服务器从 stdin 读取或客户端从 stdout 读取）需要不断地读取数据，直到遇到一个换行符 (\n)。从开始或上一个换行符到当前换行符之间的字节流（不包括换行符本身）就应该是一个完整的 JSON-RPC 消息。
  • 日志输出 (stderr):服务器进程可以使用其 stderr 通道输出日志信息。这纯粹是可选的 (MAY)。日志内容应该是人类或机器可读的 UTF-8 编码字符串，通常包含调试信息、状态报告、错误详情等。
  • 服务器不得向其写入任何stdout无效的 MCP 消息;客户端不得向服务器写入任何stdin无效的 MCP 消息。

2.1 Streamable HTTP(可流式传输的 HTTP)

详细解释

MCP（Model Context Protocol）服务器在“Streamable HTTP transport”模式下的工作方式：

  1. 独立进程：服务器作为独立进程运行，可以同时处理多个客户端连接。
  2. HTTP POST 和 GET：客户端通过 HTTP 的 POST 和 GET 方法与服务器交互。
  3. SSE（Server-Sent Events）可选：服务器可以选择使用 SSE 技术，实现消息流式推送（如实时通知、流式响应）。
  4. 单一 HTTP 端点：服务器只需暴露一个 HTTP 路径（如 /mcp），同时支持 POST 和 GET 请求。
  5. 灵活性：既支持基础的 MCP 功能，也支持更高级的流式和通知功能。

服务器还用上面给出的，对应着我们写一个客户端：

import asyncio  # 导入异步编程库
from mcp.client.session import ClientSession  # 导入 MCP 客户端会话类
from mcp.client.sse import sse_client  # 导入用于 SSE（服务端推送）连接的客户端async def main():# 使用 async with 正确管理异步上下文async with sse_client(url="http://127.0.0.1:8000/sse") as (read_stream, write_stream):# 建立 SSE 连接，获取读写流async with ClientSession(read_stream, write_stream) as session:# 创建 MCP 会话对象，并进入异步上下文await session.initialize()# 初始化会话，准备与服务端交互prompts = (await session.list_prompts()).prompts# 异步获取所有可用的 prompt（提示函数）列表for prompt in prompts:# 遍历每个 promptprint(f"Prompt name: {prompt.name}, title: {prompt.title}, description: {prompt.description}")# 打印 prompt 的名称、标题和描述if __name__ == '__main__':asyncio.run(main())# 如果作为主程序运行，则启动异步主函数

先启动服务器，再启动客户端就可以获取如下结果、
在这里插入图片描述

2 Images

FastMCP提供了一个自动处理图像数据的Image类：
服务端：

"""Example showing image handling with FastMCP."""
# 示例：演示如何用 FastMCP 处理图片from PIL import Image as PILImage  # 导入 Pillow 库中的 Image 类，并重命名为 PILImage
from mcp.server.fastmcp import FastMCP, Image  # 导入 FastMCP 框架的主类和图片类型
import io
mcp = FastMCP("Image Example")  # 创建一个 FastMCP 服务实例，名称为 "Image Example"@mcp.tool()  # 注册一个工具函数，使其可以被 FastMCP 服务调用
def create_thumbnail(image_path: str) -> Image:"""Create a thumbnail from an image"""# 创建缩略图：输入参数为图片路径，返回值为 Image 类型img = PILImage.open(image_path)  # 打开指定路径的图片文件img_byte_arr = io.BytesIO()img.save(img_byte_arr, format='PNG')img.thumbnail((100, 100))        # 生成最大尺寸为 100x100 的缩略图return Image(data=img_byte_arr.getvalue(), format="png")  # 将图片数据转换为字节流，并以 PNG 格式返回if __name__ == '__main__':mcp.run(transport='sse')

客户端：

import asyncio
from PIL import Image as PILImage
import io
import base64
from mcp.client.session import ClientSession
from mcp.client.sse import sse_clientasync def main():# 连接到服务端 SSEasync with sse_client(url="http://127.0.0.1:8000/sse") as (read_stream, write_stream):async with ClientSession(read_stream, write_stream) as session:await session.initialize()# 调用 create_thumbnail 工具，传入图片路径result = await session.call_tool(你的函数名，你的图片位置)# 输出图片字节流img_bytes = base64.b64decode(result.content[0].data)print("图片字节流:", img_bytes)# 用 PIL 还原并显示图片img = PILImage.open(io.BytesIO(img_bytes))img.show()  # 在本地弹出图片窗口if __name__ == '__main__':asyncio.run(main())

3 Context

Context 在 FastMCP 框架中用于工具函数与服务端交互，主要作用如下：

  1. 发送信息：可以通过 ctx.info()、ctx.debug() 等方法向客户端发送实时消息或日志。
  2. 报告进度：用 ctx.report_progress() 方法向客户端推送任务进度，便于前端显示进度条或状态。
  3. 管理会话状态：Context 可以保存和传递会话相关的数据，实现更复杂的交互逻辑。
  4. 异步通信：支持异步操作，适合长时间运行的任务，保证服务端和客户端实时同步状态。

简而言之，Context 是工具函数与客户端之间的桥梁，负责消息、进度和状态的实时推送与管理。

服务端：

from mcp.server.fastmcp import Context, FastMCP  # 导入 FastMCP 框架的上下文和主类mcp = FastMCP(name="Progress Example")  # 创建一个 FastMCP 服务实例，名称为 "Progress Example"@mcp.tool()  # 注册一个工具函数，使其可以被 FastMCP 服务调用
async def long_running_task(task_name: str, ctx: Context, steps: int = 5) -> str:"""Execute a task with progress updates."""# 定义一个异步长任务工具，支持进度更新。参数包括任务名、上下文对象和步骤数。await ctx.info(f"Starting: {task_name}")# 通过上下文发送任务开始的信息for i in range(steps):progress = (i + 1) / steps# 计算当前进度（0~1之间的小数）await ctx.report_progress(progress=progress,total=1.0,message=f"Step {i + 1}/{steps}",)# 通过上下文报告当前进度和提示信息await ctx.debug(f"Completed step {i + 1}")# 发送调试信息，说明当前步骤已完成return f"Task '{task_name}' completed"# 所有步骤完成后，返回任务完成的提示字符串if __name__ == '__main__':mcp.run(transport='sse')# 如果作为主程序运行，则启动 FastMCP 服务，使用 SSE 作为通信方式
http://www.lryc.cn/news/589899.html

相关文章：

  • 项目学习笔记 display从none切换成block
  • 尚庭公寓-------图片上传接口
  • 2025年工会考试题库及答案
  • alpineLinux修改包管理为国内源
  • 详解SPFA算法-单源最短路径求解
  • 陆面、生态、水文模拟与多源遥感数据同化的实践技术应用
  • 【图灵完备】算数运算
  • sktime - 时间序列机器学习统一接口
  • 控制Vue对话框显示隐藏
  • C++设计模式之创建型模式
  • 【机器学习】数据理解：数据导入、数据审查与数据可视化
  • 数据降维方法：PCA
  • 集训Day02笔记总结（关于一些OJ题目的）
  • 第四章 OB SQL调优
  • Taro.eventCenter 用法详解与实战
  • DAY8-在地下城寻求邂逅Python是否搞错了什么
  • JavaScript语言 Error对象及错误处理机制 原生错误类型
  • Matlab数字图像处理——基于图像分割与模板匹配的的车牌识别系统
  • orfeotoolbox ResetMargin
  • mongoDB初始化项目简单操作示例
  • Windows 启动后桌面黑屏，其他程序正常运行
  • ARCGIS PRO DSK 颜色选择控件（ColorPickerControl）的调用
  • MySQL 8.0 OCP 1Z0-908 题目解析(28)
  • 数据库（five day）——物物而不物于物，念念而不念于念。
  • JAVA面试宝典 -《分布式ID生成器：Snowflake优化变种》
  • uniapp+vue3+鸿蒙系统的开发
  • Rust入门之并发编程基础（三）
  • Android开发知识点总结合集
  • 基于docker的redis集群
  • 智慧公厕系统打造洁净、安全的公共空间