1.提示词 #

• 什么是提示词？

本节将介绍在 FastMCP 协议体系下何为“Prompt”，它的基本用法、接口结构，以及如何利用 API 进行 prompt 的注册、获取和测试。你将看到 prompt 在多轮对话和工具自动化中的价值，并了解如何通过 mcp_lite 客户端与服务端对 prompt 进行交互。在实际开发中，prompt 用于引导模型的回复，让工具和模型更好地协作。

主要内容包括：

• Prompt 的定义及用途
• Prompt 的基本数据结构（如 PromptMessage、GetPromptResult 等）
• 服务端提供的 prompts/list 和 prompts/get 接口功能
• 客户端如何通过 list_prompts、get_prompt 等方法获取和测试 prompt
• 典型场景与工作流举例

通过本节内容，你可以快速上手 prompt 的注册、调用流程，并结合实际业务需求灵活配置、复用 prompt 资源。

npx @modelcontextprotocol/inspector uv --directory D:/forever/docs/mcpsdk2 run prompts_server.py

2. init.py #

mcp_lite/server/init.py

# mcp_lite.server 包

3. init.py #

mcp_lite/server/fastmcp/init.py

# FastMCP 包：从原 fastmcp 模块导入所有内容以保持向后兼容
# 原 fastmcp.py 的内容已移入此包，通过 __init__ 导出

# 导入系统模块
import sys
# 导入 base64 编码库
import base64
# 导入 json 序列化
import json
# 导入正则表达式模块
import re
# 导入 url 解码工具
from urllib.parse import unquote
# 导入类型相关辅助函数
from typing import get_args, get_origin, get_type_hints
# 导入 SessionMessage 类型
from mcp_lite.message import SessionMessage
# 导入 stdio 通信模块
from mcp_lite.server import stdio
# 导入函数签名和异步工具
import inspect
import asyncio
from mcp_lite.types import (                                       # 导入 mcp_lite.types 模块中的多个类型
    JSONRPCRequest,                                                # JSONRPC 请求类型
    JSONRPCError,                                                  # JSONRPC 错误响应类型
    ErrorData,                                                     # 错误数据类型
    InitializeResult,                                              # 初始化结果类型
    LATEST_PROTOCOL_VERSION,                                       # 最新协议版本号
    ToolsCapability,                                               # 工具相关能力描述类型
    ResourcesCapability,                                           # 资源相关能力描述类型
    PromptsCapability,                                             # Prompt 相关能力描述类型
    ServerCapabilities,                                            # 服务器能力类型
    Implementation,                                                # 实现信息类型
    JSONRPCResponse,                                               # JSONRPC 响应类型
    ListToolsResult,                                               # 列出工具结果类型
    Tool,                                                          # 单个工具类型
    TextContent,                                                   # 文本内容类型
    CallToolRequestParams,                                         # 调用工具请求参数类型
    CallToolResult,                                                # 调用工具响应结果类型
    Resource,                                                      # 静态资源类型
    ResourceTemplate,                                              # 资源模板类型
    ListResourcesResult,                                           # 资源列表返回结果类型
    ListResourceTemplatesResult,                                   # 资源模板列表返回结果类型
    ReadResourceRequestParams,                                     # 读取资源请求参数类型
    ReadResourceResult,                                            # 读取资源请求响应类型
    TextResourceContents,                                          # 资源文本内容类型
    BlobResourceContents,                                          # 资源二进制内容类型
    Prompt,                                                        # Prompt 类型
    PromptArgument,                                                # Prompt 参数类型
    ListPromptsResult,                                             # Prompt 列表结果类型
    GetPromptRequestParams,                                        # 获取 Prompt 请求参数类型
    GetPromptResult,                                               # 获取 Prompt 响应结果类型
    PromptMessage,                                                 # Prompt 消息类型
)
# 导入 pydantic 的基础模型作为类型判定
from pydantic import BaseModel as PydanticBaseModel
# 导入 Typeddict 测试
from typing_extensions import is_typeddict as _is_typeddict

# 导入本包下 prompts 子模块
+from . import prompts

# 辅助函数：提取函数输出 schema 及包裹需求
def _output_schema_and_wrap(fn, structured_output: bool):
    # 若不是结构化输出，直接返回
    if not structured_output:
        return None, False
    try:
        # 获取函数签名
        sig = inspect.signature(fn)
        # 获取返回类型注解
        ann = sig.return_annotation
        # 如果没有注解，返回
        if ann is inspect.Parameter.empty:
            return None, False
    except Exception:
        return None, False
    # 生成 schema
    return _schema_from_annotation(ann, fn.__name__)

# 辅助函数：根据注解和函数名生成 schema
def _schema_from_annotation(ann, func_name: str):
    # 没有注解直接返回
    if ann is inspect.Parameter.empty:
        return None, False
    # 获取注解的原类型
    origin = get_origin(ann)
    wrap = False
    schema = None
    # 如果是 Pydantic 的模型子类
    if PydanticBaseModel and isinstance(ann, type) and issubclass(ann, PydanticBaseModel):
        schema = ann.model_json_schema()
        return schema, False
    # 如果是 Typeddict
    if hasattr(ann, "__annotations__") and not (origin is dict or origin is list):
        if _is_typeddict(ann):
            hints = get_type_hints(ann) if hasattr(ann, "__annotations__") else {}
            props = {}
            for k, v in hints.items():
                t = v
                if t is int or t is type(None) and int in get_args(v):
                    props[k] = {"type": "integer"}
                elif t is float:
                    props[k] = {"type": "number"}
                elif t is str:
                    props[k] = {"type": "string"}
                elif t is bool:
                    props[k] = {"type": "boolean"}
                else:
                    props[k] = {"type": "string"}
            schema = {"type": "object", "properties": props, "required": list(props)}
            return schema, False
        try:
            hints = get_type_hints(ann)
            if hints:
                props = {}
                for k, v in hints.items():
                    if v is int or v is type(None):
                        props[k] = {"type": "integer"}
                    elif v is float:
                        props[k] = {"type": "number"}
                    elif v is str:
                        props[k] = {"type": "string"}
                    elif v is bool:
                        props[k] = {"type": "boolean"}
                    elif get_origin(v) is list:
                        props[k] = {"type": "array", "items": {"type": "string"}}
                    else:
                        props[k] = {"type": "string"}
                schema = {"type": "object", "properties": props}
                return schema, False
        except Exception:
            pass
    # 如果是 dict 类型
    if origin is dict:
        args = get_args(ann)
        if len(args) == 2 and args[0] is str:
            vt = args[1]
            # 判断 value 类型
            if vt is float:
                schema = {"type": "object", "additionalProperties": {"type": "number"}}
            elif vt is int:
                schema = {"type": "object", "additionalProperties": {"type": "integer"}}
            elif vt is str:
                schema = {"type": "object", "additionalProperties": {"type": "string"}}
            else:
                schema = {"type": "object", "additionalProperties": {}}
            return schema, False
        wrap = True
    # 如果是 list、基础类型等需要包裹
    if origin is list or ann in (int, float, str, bool, type(None)):
        wrap = True
    # 包裹输出情况
    if wrap:
        result_schema = {"type": "string"}
        if ann is int:
            result_schema = {"type": "integer"}
        elif ann is float:
            result_schema = {"type": "number"}
        elif ann is bool:
            result_schema = {"type": "boolean"}
        elif origin is list:
            result_schema = {"type": "array", "items": {"type": "string"}}
        schema = {"type": "object", "properties": {"result": result_schema}, "required": ["result"]}
        return schema, True
    return None, False

# 辅助：对象输出转换为结构化内容
def _to_structured(out, output_schema, wrap_output):
    if output_schema is None:
        return None
    try:
        # 如果是 Pydantic 模型
        if PydanticBaseModel and isinstance(out, PydanticBaseModel):
            return out.model_dump(mode="json")
        # 如果需要包裹
        if wrap_output:
            return {"result": out}
        # 如果是字典
        if isinstance(out, dict):
            return dict(out)
        # 带有 __annotations__ 的对象，提取属性
        if hasattr(out, "__annotations__"):
            hints = get_type_hints(type(out)) if hasattr(type(out), "__annotations__") else getattr(out, "__annotations__", {})
            return {k: getattr(out, k) for k in hints if hasattr(out, k)}
        # 其它直接包裹
        return {"result": out}
    except Exception:
        # 发生异常返回字符串
        return {"result": str(out)}

# 根据函数参数生成 schema
def _schema(fn):
    sig = inspect.signature(fn)
    props = {}
    req = []
    for n, p in sig.parameters.items():
        # 跳过以下划线开头的参数
        if n.startswith("_"):
            continue
        # 参数类型为字符串，标题为参数名
        props[n] = {"type": "string", "title": n}
        # 必填参数加入 required
        if p.default is inspect.Parameter.empty:
            req.append(n)
    return {"type": "object", "properties": props, "required": req}

# 根据 prompt 函数生成 prompt schema
+def _prompt_schema(fn):
+    """根据函数签名生成 Prompt 参数 schema。"""
+    sig = inspect.signature(fn)
+    props = {}
+    req = []
+    for n, p in sig.parameters.items():
+        if n.startswith("_"):
+            continue
+        props[n] = {"type": "string", "title": n}
+        if p.default is inspect.Parameter.empty:
+            req.append(n)
+    return {"type": "object", "properties": props, "required": req}

# 工具函数封装类
class _Tool:
    def __init__(self, fn, name=None, desc=None, structured_output=True):
        # 函数本体
        self.fn = fn
        # 名称
        self.name = name or fn.__name__
        # 描述
        self.desc = desc or (fn.__doc__ or "").strip()
        # 入参 schema
        self.schema = _schema(fn)
        # 是否是异步函数
        self.async_fn = asyncio.iscoroutinefunction(fn)
        # 是否结构化输出
        self.structured_output = structured_output
        # 输出 schema 及需否包裹
        self.output_schema, self.wrap_output = _output_schema_and_wrap(fn, structured_output) if structured_output else (None, False)

    # 转换为 Tool 对象
    def to_tool(self):
        return Tool(name=self.name, description=self.desc or None, input_schema=self.schema, output_schema=self.output_schema)

    # 执行工具（自动支持异步）
    def run(self, args):
        if self.async_fn:
            return asyncio.run(self.fn(**args))
        return self.fn(**args)

# 静态资源封装类
class _Resource:
    def __init__(self, uri, fn, mime_type="text/plain"):
        # 资源 URI
        self.uri = uri
        # 回调函数
        self.fn = fn
        # MIME 类型
        self.mime_type = mime_type
        # 是否异步
        self.async_fn = asyncio.iscoroutinefunction(fn)

    # 资源运行，自动支持异步
    def run(self):
        if self.async_fn:
            return asyncio.run(self.fn())
        return self.fn()

# 资源模板封装类（支持 URI 参数模板）
class _ResourceTemplate:
    def __init__(self, uri_template, fn, mime_type="text/plain"):
        # URI 模板
        self.uri_template = uri_template
        # 生成资源内容的函数
        self.fn = fn
        # MIME 类型
        self.mime_type = mime_type
        # 是否异步
        self.async_fn = asyncio.iscoroutinefunction(fn)
        # 提取函数里除了 _ 开头之外的参数名
        sig = inspect.signature(fn)
        self.param_names = [n for n in sig.parameters if not n.startswith("_")]

    # 检查 URI 是否与模板匹配，并解析参数
    def matches(self, uri):
        pattern = self.uri_template.replace("{", "(?P<").replace("}", ">[^/]+)")
        m = re.match(f"^{pattern}$", uri)
        if m:
            return {k: unquote(v) for k, v in m.groupdict().items()}
        return None

    # 执行模板内容生成函数
    def run(self, args):
        if self.async_fn:
            return asyncio.run(self.fn(**args))
        return self.fn(**args)

# Prompt 封装
+class _Prompt:
+    """内部 Prompt 封装类。"""
+
+    def __init__(self, fn, name=None, title=None, description=None):
+        # prompt 生成函数
+        self.fn = fn
+        # prompt 名称
+        self.name = name or fn.__name__
+        # prompt 标题
+        self.title = title
+        # prompt 描述
+        self.description = description or (fn.__doc__ or "").strip()
+        # 通用 schema
+        self.schema = _prompt_schema(fn)
+        # 是否异步
+        self.async_fn = asyncio.iscoroutinefunction(fn)
+        # 从 schema 生成 PromptArgument 列表
+        args_list = []
+        if "properties" in self.schema:
+            required = set(self.schema.get("required", []))
+            for pname, pinfo in self.schema["properties"].items():
+                args_list.append(
+                    PromptArgument(
+                        name=pname,
+                        description=pinfo.get("title"),
+                        required=pname in required,
+                    )
+                )
+        self.arguments = args_list
+
+    # 转成 Prompt 对象
+    def to_prompt(self):
+        return Prompt(
+            name=self.name,
+            title=self.title,
+            description=self.description or None,
+            arguments=self.arguments if self.arguments else None,
+        )
+
+    # prompt 执行
+    def run(self, args):
+        if self.async_fn:
+            return asyncio.run(self.fn(**args))
+        return self.fn(**args)

# FastMCP 主类
class FastMCP:
    def __init__(self, name="mcp-server"):
        # 服务器名称
        self.name = name
        # 工具注册表
        self._tools = {}
        # 静态资源注册表
        self._resources = {}
        # 资源模板注册表
        self._resource_templates = {}
        # prompt 注册表
+       self._prompts = {}

    # 注册工具的装饰器
    def tool(self, name=None, description=None, structured_output=True):
        def deco(fn):
            t = _Tool(fn, name, description, structured_output=structured_output)
            self._tools[t.name] = t
            return fn
        return deco

    # 注册资源或模板资源的装饰器
    def resource(self, uri, mime_type="text/plain"):
        def deco(fn):
            # 判断带模板参数还是静态资源
            if "{" in uri and "}" in uri:
                template = _ResourceTemplate(uri, fn, mime_type)
                self._resource_templates[uri] = template
            else:
                res = _Resource(uri, fn, mime_type)
                self._resources[uri] = res
            return fn
        return deco

+   # 注册 Prompt 的装饰器
+   def prompt(self, name=None, title=None, description=None):
+       """注册 Prompt 的装饰器。"""
+       def deco(fn):
+           p = _Prompt(fn, name=name, title=title, description=description)
+           self._prompts[p.name] = p
+           return fn
+       return deco

    # 核心 RPC 方法分发
    def _handle(self, req):
        # 获取方法名、参数和请求 id
        method, params, rid = req.method, req.params or {}, req.id
        # 初始化请求
        if method == "initialize":
            caps = ServerCapabilities(
                tools=ToolsCapability(),
                resources=ResourcesCapability(),
+               prompts=PromptsCapability() if self._prompts else None,
            )
            r = InitializeResult(
                protocol_version=LATEST_PROTOCOL_VERSION,
                capabilities=caps,
                server_info=Implementation(name=self.name, version="0.1.0"),
            )
            return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))
        # 工具列表请求
        if method == "tools/list":
            r = ListToolsResult(tools=[t.to_tool() for t in self._tools.values()])
            return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))
        # 工具调用请求
        if method == "tools/call":
            p = CallToolRequestParams.model_validate(params, by_name=False)
            t = self._tools.get(p.name)
            if not t:
                return JSONRPCError(
                    jsonrpc="2.0",
                    id=rid,
                    error=ErrorData(code=-32602, message=f"Unknown tool: {p.name}")
                )
            try:
                out = t.run(p.arguments or {})
                # 如果直接返回的是 CallToolResult
                if isinstance(out, CallToolResult):
                    r = out
                else:
                    struct = None
                    # 结构化输出
                    if t.output_schema is not None:
                        struct = _to_structured(out, t.output_schema, t.wrap_output)
                    # 字符串直接用文本包装
                    if isinstance(out, str):
                        c = [TextContent(text=out)]
                    elif out is None:
                        c = []
                    elif struct is not None:
                        # 结构化结果按 JSON 格式化后返回
                        c = [TextContent(text=json.dumps(struct, ensure_ascii=False, indent=2))]
                    else:
                        # 普通对象尝试转 json，否则转字符串
                        try:
                            if PydanticBaseModel and isinstance(out, PydanticBaseModel):
                                text = json.dumps(out.model_dump(mode="json"), ensure_ascii=False, indent=2)
                            elif isinstance(out, dict):
                                text = json.dumps(out, ensure_ascii=False, indent=2)
                            else:
                                text = str(out)
                        except Exception:
                            text = str(out)
                        c = [TextContent(text=text)]
                    r = CallToolResult(content=c, structured_content=struct)
            except Exception as e:
                r = CallToolResult(content=[TextContent(text=str(e))], is_error=True)
            return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))

        # prompts/list 请求
+       if method == "prompts/list":
+           prompts_list = [p.to_prompt() for p in self._prompts.values()]
+           r = ListPromptsResult(prompts=prompts_list)
+           return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))

        # prompts/get 请求
+       if method == "prompts/get":
+           p = GetPromptRequestParams.model_validate(params, by_name=False)
+           prompt_obj = self._prompts.get(p.name)
+           if not prompt_obj:
+               return JSONRPCError(
+                   jsonrpc="2.0",
+                   id=rid,
+                   error=ErrorData(code=-32602, message=f"Unknown prompt: {p.name}")
+               )
+           try:
+               args = p.arguments or {}
+               result = prompt_obj.run(args)
+               messages = []
+               # 单条消息包装为列表
+               if not isinstance(result, (list, tuple)):
+                   result = [result]
+               for msg in result:
+                   # 若为内置的 Message 对象
+                   if isinstance(msg, prompts.base.Message):
+                       content = msg.content
+                       if isinstance(content, str):
+                           content = TextContent(type="text", text=content)
+                       messages.append(PromptMessage(role=msg.role, content=content))
+                   # 字符串消息按 user 角色组装
+                   elif isinstance(msg, str):
+                       messages.append(PromptMessage(role="user", content=TextContent(type="text", text=msg)))
+                   # dict 消息，读 role 和 content
+                   elif isinstance(msg, dict):
+                       role = msg.get("role", "user")
+                       cnt = msg.get("content", "")
+                       if isinstance(cnt, dict) and cnt.get("type") == "text":
+                           content = TextContent(**cnt)
+                       else:
+                           content = TextContent(type="text", text=str(cnt) if not isinstance(cnt, str) else cnt)
+                       messages.append(PromptMessage(role=role, content=content))
+                   # 其它均归为 user 文本
+                   else:
+                       messages.append(PromptMessage(role="user", content=TextContent(type="text", text=str(msg))))
+               r = GetPromptResult(description=prompt_obj.description, messages=messages)
+           except Exception as e:
+               return JSONRPCError(jsonrpc="2.0", id=rid, error=ErrorData(code=-32603, message=str(e)))
+           return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))

        # 资源静态列表
        if method == "resources/list":
            resources = [
                Resource(uri=u, name=u, mime_type=r.mime_type)
                for u, r in self._resources.items()
            ]
            r = ListResourcesResult(resources=resources)
            return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))

        # 资源模板列表
        if method == "resources/templates/list":
            templates = [
                ResourceTemplate(uri_template=u, name=u, mime_type=t.mime_type)
                for u, t in self._resource_templates.items()
            ]
            r = ListResourceTemplatesResult(resource_templates=templates)
            return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))

        # 读取资源接口
        if method == "resources/read":
            p = ReadResourceRequestParams.model_validate(params, by_name=False)
            uri_str = str(p.uri)
            try:
                # 静态资源优先
                if res := self._resources.get(uri_str):
                    out = res.run()
                    content = TextResourceContents(uri=uri_str, text=str(out), mime_type=res.mime_type)
                    r = ReadResourceResult(contents=[content])
                    return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))
                # 匹配模板资源
                for template in self._resource_templates.values():
                    if args := template.matches(uri_str):
                        out = template.run(args)
                        if isinstance(out, bytes):
                            content = BlobResourceContents(
                                uri=uri_str,
                                blob=base64.b64encode(out).decode(),
                                mime_type=template.mime_type or "application/octet-stream",
                            )
                        else:
                            content = TextResourceContents(
                                uri=uri_str,
                                text=str(out),
                                mime_type=template.mime_type or "text/plain",
                            )
                        r = ReadResourceResult(contents=[content])
                        return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))
                # 找不到资源
                return JSONRPCError(jsonrpc="2.0", id=rid, error=ErrorData(code=-32602, message=f"Unknown resource: {uri_str}"))
            except Exception as e:
                return JSONRPCError(jsonrpc="2.0", id=rid, error=ErrorData(code=-32603, message=str(e)))

        # 未知方法
        return JSONRPCError(jsonrpc="2.0", id=rid, error=ErrorData(code=-32601, message=f"Method not found: {method}"))

    # 消息包装与请求
    def _handle_msg(self, msg):
        # 非 SessionMessage 类型忽略
        if not isinstance(msg, SessionMessage):
            return None
        m = msg.message
        # 必须是带 id 的 jsonrpc 请求
        if not isinstance(m, JSONRPCRequest) or getattr(m, "id", None) is None:
            return None
        #print("[Server] Request:", m.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
        try:
            resp = self._handle(m)
            #print("[Server] Response:", resp.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
            return resp
        except Exception as e:
            err = JSONRPCError(jsonrpc="2.0", id=m.id, error=ErrorData(code=-32603, message=str(e)))
            #print("[Server] Response (error):", err.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
            return err

    # 运行服务器主逻辑
    def run(self, transport="stdio"):
        # 目前只支持 stdio
        if transport != "stdio":
            raise ValueError(f"unsupported transport: {transport}")
        # 启动 stdio 服务器
        stdio.stdio_server(self._handle_msg)

# 明确导出接口
+__all__ = ["FastMCP", "prompts"]

4. init.py #

mcp_lite/server/fastmcp/prompts/init.py

# prompts 子包，提供 Prompt 相关消息类型
from . import base

__all__ = ["base"]

5. base.py #

mcp_lite/server/fastmcp/prompts/base.py

# 基础消息类型，用于 Prompt 返回的消息列表
from __future__ import annotations

# 导入 Any 和 Literal 类型，用于类型提示
from typing import Any, Literal

# 导入 Pydantic 的基类 BaseModel
from pydantic import BaseModel

# 导入自定义的 TextContent 类型
from mcp_lite.types import TextContent

# 定义消息基类，继承自 Pydantic 的 BaseModel
class Message(BaseModel):
    """Prompt 消息基类。"""

    # 消息的角色，只允许为 "user" 或 "assistant"
    role: Literal["user", "assistant"]
    # 消息的内容，可以是 TextContent 或字典
    content: TextContent | dict[str, Any]

    # 初始化方法，允许 content 为 str、TextContent 或 dict
    def __init__(self, content: str | TextContent | dict[str, Any], role: str = "user", **kwargs: Any):
        # 如果 content 是字符串，则封装为 TextContent 对象
        if isinstance(content, str):
            content = TextContent(type="text", text=content)
        # 调用父类的初始化方法，完成属性设置
        super().__init__(content=content, role=role, **kwargs)

# 用户消息类，默认 role 为 "user"，继承于 Message
class UserMessage(Message):
    """用户消息。"""

    # 角色属性，默认为 "user"
    role: Literal["user", "assistant"] = "user"

    # 初始化方法，强制 role 为 "user"
    def __init__(self, content: str | TextContent | dict[str, Any], **kwargs: Any):
        super().__init__(content=content, role="user", **kwargs)

# 助手消息类，默认 role 为 "assistant"，继承于 Message
class AssistantMessage(Message):
    """助手消息。"""

    # 角色属性，默认为 "assistant"
    role: Literal["user", "assistant"] = "assistant"

    # 初始化方法，强制 role 为 "assistant"
    def __init__(self, content: str | TextContent | dict[str, Any], **kwargs: Any):
        super().__init__(content=content, role="assistant", **kwargs)

6. prompts_client.py #

prompts_client.py

# 导入os库，用于处理文件和路径操作
import os

# 从mcp_lite包导入客户端会话、Stdio服务器参数和类型定义
from mcp_lite import ClientSession, StdioServerParameters, types

# 从mcp_lite.client.stdio导入stdio_client工厂函数
from mcp_lite.client.stdio import stdio_client

# 定义辅助函数，用于打印Prompt消息内容
def print_prompt_messages(tag: str, prompt: types.GetPromptResult) -> None:
    # 创建一个空列表用于保存格式化后的消息行
    lines: list[str] = []
    # 遍历prompt中的每条消息
    for msg in prompt.messages:
        # 获取消息的角色
        role = msg.role
        # 获取消息内容
        content = msg.content
        # 如果内容是文本类型
        if isinstance(content, types.TextContent):
            # 将角色和文本内容格式化加入列表
            lines.append(f"({role}) {content.text}")
        else:
            # 对于非文本内容，记录其类型名称
            lines.append(f"({role}) <非文本内容：{type(content).__name__}>")
    # 打印所有整理后的消息
    print(f"[{tag}]\n" + "\n".join(lines))

# 主函数，脚本执行入口
def main() -> None:
    # 获取当前文件所在目录
    base_dir = os.path.dirname(os.path.abspath(__file__))
    # 拼接Prompts服务器的脚本路径
    server_path = os.path.join(base_dir, "prompts_server.py")
    # 构造Stdio服务器参数
    server_params = StdioServerParameters(
        command="python",         # 启动命令为python
        args=[server_path],       # 传入Prompts服务器脚本路径
        env={},                   # 不传递额外环境变量
    )
    # 使用with语句启动服务器并获取读写通道
    with stdio_client(server_params) as (read, write):
        # 创建客户端会话
        session = ClientSession(read, write)
        # 初始化会话，完成握手
        session.initialize()
        # 获取所有Prompt列表
        prompts = session.list_prompts()
        # 打印所有Prompt的名称
        print("[Prompts]", [p.name for p in prompts.prompts])
        # 调用greet_user Prompt，传入参数
        prompt_greet = session.get_prompt(
            "greet_user", arguments={"name": "Alice", "style": "friendly"}
        )
        # 打印greet_user prompt返回的消息
        print_prompt_messages("greet_user", prompt_greet)
        # 调用debug_assistant Prompt，传入错误参数
        prompt_debug = session.get_prompt(
            "debug_assistant",
            arguments={"error": "IndexError: list index out of range"},
        )
        # 打印debug_assistant prompt返回的消息
        print_prompt_messages("debug_assistant", prompt_debug)

# 脚本主入口判断
if __name__ == "__main__":
    # 调用主函数
    main()

官方代码

# 导入os库，用于处理文件和路径操作
import os
import asyncio
# 从mcp_lite包导入客户端会话、Stdio服务器参数和类型定义
from mcp import ClientSession, StdioServerParameters, types

# 从mcp_lite.client.stdio导入stdio_client工厂函数
from mcp.client.stdio import stdio_client

# 定义辅助函数，用于打印Prompt消息内容
def print_prompt_messages(tag: str, prompt: types.GetPromptResult) -> None:
    # 创建一个空列表用于保存格式化后的消息行
    lines: list[str] = []
    # 遍历prompt中的每条消息
    for msg in prompt.messages:
        # 获取消息的角色
        role = msg.role
        # 获取消息内容
        content = msg.content
        # 如果内容是文本类型
        if isinstance(content, types.TextContent):
            # 将角色和文本内容格式化加入列表
            lines.append(f"({role}) {content.text}")
        else:
            # 对于非文本内容，记录其类型名称
            lines.append(f"({role}) <非文本内容：{type(content).__name__}>")
    # 打印所有整理后的消息
    print(f"[{tag}]\n" + "\n".join(lines))

# 主函数，脚本执行入口
async def main() -> None:
    # 获取当前文件所在目录
    base_dir = os.path.dirname(os.path.abspath(__file__))
    # 拼接Prompts服务器的脚本路径
    server_path = os.path.join(base_dir, "prompts_server.py")
    # 构造Stdio服务器参数
    server_params = StdioServerParameters(
        command="python",         # 启动命令为python
        args=[server_path],       # 传入Prompts服务器脚本路径
        env={},                   # 不传递额外环境变量
    )
    # 使用with语句启动服务器并获取读写通道
    async with stdio_client(server_params) as (read, write):
        # 创建客户端会话
        async with ClientSession(read, write) as session:
            # 初始化会话，完成握手
            await session.initialize()
            # 获取所有Prompt列表
            prompts = await session.list_prompts()
            # 打印所有Prompt的名称
            print("[Prompts]", [p.name for p in prompts.prompts])
            # 调用greet_user Prompt，传入参数
            prompt_greet = await session.get_prompt(
                "greet_user", arguments={"name": "Alice", "style": "friendly"}
            )
            # 打印greet_user prompt返回的消息
            print_prompt_messages("greet_user", prompt_greet)
            # 调用debug_assistant Prompt，传入错误参数
            prompt_debug = await session.get_prompt(
                "debug_assistant",
                arguments={"error": "IndexError: list index out of range"},
            )
            # 打印debug_assistant prompt返回的消息
            print_prompt_messages("debug_assistant", prompt_debug)

# 脚本主入口判断
if __name__ == "__main__":
    # 调用主函数
    asyncio.run(main())

7. prompts_server.py #

prompts_server.py

# 导入 sys 模块，用于设置标准流编码
import sys
# 导入 FastMCP 高层服务器
from mcp_lite.server.fastmcp import FastMCP

# 导入基础消息类型（如 UserMessage、AssistantMessage 等）
from mcp_lite.server.fastmcp.prompts import base

# 创建 FastMCP 服务器实例，名称为 "Prompts Server"，客户端可见
mcp = FastMCP(name="Prompts Server")

# 一、定义返回字符串的 Prompt（最简单形式）

# 使用装饰器将函数注册为 Prompt，函数参数会自动变为可配置参数
@mcp.prompt()
def greet_user(name: str, style: str = "friendly") -> str:
    # 提供文档字符串，说明该函数的功能
    """根据姓名与风格生成问候提示词（字符串形式）。"""
    # 定义不同风格的提示语模板
    styles: dict[str, str] = {
        "friendly": "请用温暖、友好的语气写一段问候",
        "formal": "请用正式、专业的语气写一段问候",
        "casual": "请用轻松、随意的语气写一段问候",
    }
    # 根据传入的 style 参数选择对应风格，拼接用户名，返回最终提示字符串
    return f"{styles.get(style, styles['friendly'])}，对象是名为 {name} 的用户。"

# 二、定义返回消息列表的 Prompt（更灵活，适合多轮引导）

# 使用装饰器注册为 Prompt，并设置显示标题为“调试助手”
@mcp.prompt(title="调试助手")
def debug_assistant(error: str) -> list[base.Message]:
    # 提供文档字符串，说明本 Prompt 返回一组消息用于调试指导
    """返回一组消息，用于引导调试对话（消息列表形式）。"""
    # 返回一个消息列表，包含用户描述错误、补充错误详情、以及助手引导回复
    return [
        # 第一条消息，由用户发出，表明遇到错误
        base.UserMessage("我遇到了如下错误："),
        # 第二条消息，用户补充具体的错误详情
        base.UserMessage(error),
        # 第三条消息，助手请求用户进一步提供信息以协助定位问题
        base.AssistantMessage(
            "我将帮助你定位问题。请提供你已尝试的步骤、相关代码片段，以及你所期望的结果。"
        ),
    ]

# 主程序入口，判断当前模块是否为主模块
if __name__ == "__main__":
    # 判断标准输出是否支持 reconfigure 方法，并设置编码为 utf-8
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8")
        sys.stdin.reconfigure(encoding="utf-8")
    # 启动 MCP 服务器，使用 stdio 作为通信方式
    mcp.run(transport="stdio")

官方代码

# 导入 sys 模块，用于设置标准流编码
import sys
# 导入 FastMCP 高层服务器
from mcp.server.fastmcp import FastMCP

# 导入基础消息类型（如 UserMessage、AssistantMessage 等）
from mcp.server.fastmcp.prompts import base

# 创建 FastMCP 服务器实例，名称为 "Prompts Server"，客户端可见
mcp = FastMCP(name="Prompts Server")

# 一、定义返回字符串的 Prompt（最简单形式）

# 使用装饰器将函数注册为 Prompt，函数参数会自动变为可配置参数
@mcp.prompt()
def greet_user(name: str, style: str = "friendly") -> str:
    # 提供文档字符串，说明该函数的功能
    """根据姓名与风格生成问候提示词（字符串形式）。"""
    # 定义不同风格的提示语模板
    styles: dict[str, str] = {
        "friendly": "请用温暖、友好的语气写一段问候",
        "formal": "请用正式、专业的语气写一段问候",
        "casual": "请用轻松、随意的语气写一段问候",
    }
    # 根据传入的 style 参数选择对应风格，拼接用户名，返回最终提示字符串
    return f"{styles.get(style, styles['friendly'])}，对象是名为 {name} 的用户。"

# 二、定义返回消息列表的 Prompt（更灵活，适合多轮引导）

# 使用装饰器注册为 Prompt，并设置显示标题为“调试助手”
@mcp.prompt(title="调试助手")
def debug_assistant(error: str) -> list[base.Message]:
    # 提供文档字符串，说明本 Prompt 返回一组消息用于调试指导
    """返回一组消息，用于引导调试对话（消息列表形式）。"""
    # 返回一个消息列表，包含用户描述错误、补充错误详情、以及助手引导回复
    return [
        # 第一条消息，由用户发出，表明遇到错误
        base.UserMessage("我遇到了如下错误："),
        # 第二条消息，用户补充具体的错误详情
        base.UserMessage(error),
        # 第三条消息，助手请求用户进一步提供信息以协助定位问题
        base.AssistantMessage(
            "我将帮助你定位问题。请提供你已尝试的步骤、相关代码片段，以及你所期望的结果。"
        ),
    ]

# 主程序入口，判断当前模块是否为主模块
if __name__ == "__main__":
    # 判断标准输出是否支持 reconfigure 方法，并设置编码为 utf-8
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8")
        sys.stdin.reconfigure(encoding="utf-8")
    # 启动 MCP 服务器，使用 stdio 作为通信方式
    mcp.run(transport="stdio")

8. session.py #

mcp_lite/client/session.py

import sys
# 导入 SessionMessage 类
from mcp_lite.message import SessionMessage
# 从 mcp_lite.types 模块导入相关类型和常量
+from mcp_lite.types import (                                            # 从 mcp_lite.types 模块导入以下类型
+   InitializeRequestParams,                                            #   初始化请求参数类型
+   InitializeRequest,                                                  #   初始化请求类型
+   LATEST_PROTOCOL_VERSION,                                            #   最新协议版本常量
+   ClientCapabilities,                                                 #   客户端能力类型
+   Implementation,                                                     #   实现信息类型
+   InitializedNotification,                                            #   初始化完成通知类型
+   InitializeResult,                                                   #   初始化结果类型
+   JSONRPCRequest,                                                     #   JSONRPC请求类型
+   JSONRPCResponse,                                                    #   JSONRPC响应类型
+   JSONRPCError,                                                       #   JSONRPC错误类型
+   JSONRPCNotification,                                                #   JSONRPC通知类型
+   ListToolsRequest,                                                   #   工具列表请求类型
+   ListToolsResult,                                                    #   工具列表结果类型
+   CallToolResult,                                                     #   调用工具响应类型
+   CallToolRequest,                                                    #   调用工具请求类型
+   CallToolRequestParams,                                              #   调用工具请求参数类型
+   ListResourcesRequest,                                               #   资源列表请求类型
+   ListResourcesResult,                                                #   资源列表响应类型
+   ListResourceTemplatesRequest,                                       #   资源模板列表请求类型
+   ListResourceTemplatesResult,                                        #   资源模板列表响应类型
+   ReadResourceRequest,                                                #   读取资源请求类型
+   ReadResourceRequestParams,                                          #   读取资源请求参数类型
+   ReadResourceResult,                                                 #   读取资源响应类型
+   ListPromptsRequest,                                                 #   Prompt 列表请求类型
+   ListPromptsResult,                                                  #   Prompt 列表响应类型
+   GetPromptRequest,                                                   #   获取 Prompt 请求类型
+   GetPromptRequestParams,                                             #   获取 Prompt 请求参数类型
+   GetPromptResult,                                                    #   获取 Prompt 响应类型
)
# 定义 MCPError 异常类，用于表示协议相关错误
class MCPError(Exception):
    # 初始化方法，接收错误码、错误信息和可选的数据
    def __init__(self, code, message, data=None):
        # 将错误码、错误信息、附加数据赋值为实例属性
        self.code, self.message, self.data = code, message, data
        # 调用父类 Exception 的初始化，只传递错误信息
        super().__init__(message)

    # 类方法，利用 JSONRPCError 对象创建 MCPError 实例
    @classmethod
    def from_jsonrpc(cls, err):
        # 从 JSONRPCError.error 对象中取出 code/message/data 创建 MCPError
        return cls(err.error.code, err.error.message, err.error.data)
# 定义 ClientSession 类，用于描述客户端会话
class ClientSession:
    # 构造方法，接收读取流和写入流
    def __init__(self, read_stream, write_stream):
        # 赋值读取流和写入流
        self._read, self._write = read_stream, write_stream
        # 初始化请求 id 计数器为 0
        self._req_id = 0
    # 内部方法，发送请求并同步等待响应
    def _request(self, req):
        # 当前请求 id
        rid = self._req_id
        # 自增请求 id，确保下次唯一
        self._req_id += 1
        # 对请求对象进行序列化，转为 dict
        d = req.model_dump(by_alias=True, mode="json", exclude_none=True)
        # 构建 JSONRPCRequest，再封装入 SessionMessage，通过写入流发送
        jreq = JSONRPCRequest(jsonrpc="2.0", id=rid, method=d["method"], params=d.get("params"))
        # 打印请求信息到标准错误流
        #print("[Client] Request:", jreq.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
        # 将请求对象封装成 SessionMessage 后发送
        self._write.send(SessionMessage(message=jreq))
        # 循环等待响应
        while True:
            # 从读取流取出一条消息
            msg = self._read.get()
            # 若取到 None，说明连接断开，抛出异常
            if msg is None:
                raise MCPError(-32000, "Connection closed")
            # 若消息类型不是 SessionMessage，忽略继续等
            if not isinstance(msg, SessionMessage):
                continue
            # 取消息内容
            m = msg.message
            # 若为 Response 或 Error，且 id 匹配
            if isinstance(m, (JSONRPCResponse, JSONRPCError)) and getattr(m, "id", None) == rid:
                #print("[Client] Response:", m.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
                # 若为错误，抛出 MCPError 异常
                if isinstance(m, JSONRPCError):
                    raise MCPError.from_jsonrpc(m)
                # 为正常响应则返回结果数据
                return m.result
    # 内部方法，发送通知消息
    def _notify(self, n):
        # 通知对象序列化为 dict
        d = n.model_dump(by_alias=True, mode="json", exclude_none=True)
        # 封装为 JSONRPCNotification，再封装成 SessionMessage 后发出
        self._write.send(SessionMessage(
            message=JSONRPCNotification(
                jsonrpc="2.0",
                method=d["method"],
                params=d.get("params"),
            )
        ))            
    # 初始化过程，发送初始化请求，收到响应后再发初始化完成通知
    def initialize(self):
        # 构造初始化请求，并发送等待返回
        r = self._request(InitializeRequest(
            params=InitializeRequestParams(
                protocol_version=LATEST_PROTOCOL_VERSION,         # 协议版本号
                capabilities=ClientCapabilities(),                # 客户端能力
                client_info=Implementation(name="mcp_lite", version="0.1.0"),  # 客户端信息
            )
        ))
        # 发送"初始化完成"通知
        self._notify(InitializedNotification())
        # 用 InitializeResult 验证响应并返回
        return InitializeResult.model_validate(r, by_name=False)
    def list_tools(self):
        # 发送 ListToolsRequest 请求，使用 ListToolsResult 校验并返回
        return ListToolsResult.model_validate(self._request(ListToolsRequest()), by_name=False)   
    # 调用指定工具方法，传入工具名和参数
    def call_tool(self, name, arguments=None):
        # 构造 CallToolRequest 并发送，请求参数包含工具名和参数
        return CallToolResult.model_validate(
            self._request(CallToolRequest(
                params=CallToolRequestParams(name=name, arguments=arguments or {})
            )),
            by_name=False,
        )

    # 列出已注册的静态资源
    def list_resources(self):
        # 调用 _request 方法发送 ListResourcesRequest 请求
        # 使用 ListResourcesResult 的 model_validate 进行结果校验和结构化
        return ListResourcesResult.model_validate(
            self._request(ListResourcesRequest()),
            by_name=False,
        )

    # 列出资源模板（带占位符的资源）
    def list_resource_templates(self):
        # 调用 _request 方法发送 ListResourceTemplatesRequest 请求
        # 使用 ListResourceTemplatesResult 的 model_validate 进行结果校验和结构化
        return ListResourceTemplatesResult.model_validate(
            self._request(ListResourceTemplatesRequest()),
            by_name=False,
        )

    # 读取指定 URI 的资源内容
    def read_resource(self, uri):
        # uri 可为 str 或 AnyUrl 类型
        # 构造 ReadResourceRequestParams 并发送 ReadResourceRequest 请求
        # 使用 ReadResourceResult 的 model_validate 进行结果校验和结构化
        return ReadResourceResult.model_validate(
            self._request(ReadResourceRequest(params=ReadResourceRequestParams(uri=str(uri)))),
            by_name=False,
        )

    # 列出已注册的 Prompt
    # 定义 list_prompts 方法，用于获取所有已注册的 Prompt
+   def list_prompts(self):
        # 发送 ListPromptsRequest 请求，校验响应并结构化为 ListPromptsResult
+       return ListPromptsResult.model_validate(
+           self._request(ListPromptsRequest()),  # 请求所有 Prompt
+           by_name=False,                        # 按字段名校验
+       )

    # 获取指定 Prompt 的内容
    # 定义 get_prompt 方法，根据名称和参数获取指定 Prompt 的详情
+   def get_prompt(self, name, arguments=None):
        # 发送 GetPromptRequest，请求参数包含指定名称和参数，响应结构化为 GetPromptResult
+       return GetPromptResult.model_validate(
+           self._request(GetPromptRequest(                       # 发送请求
+               params=GetPromptRequestParams(                    # 构造请求参数
+                   name=name,                                    # Prompt 名称
+                   arguments=arguments or {}                     # Prompt 参数（默认为空字典）
+               )
+           )),
+           by_name=False,                                        # 按字段名校验
+       )

9. types.py #

mcp_lite/types.py

# 从 pydantic 导入 BaseModel、ConfigDict、Field、TypeAdapter、field_validator
from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, field_validator
# 导入 pydantic 的 to_camel 驼峰命名生成器
from pydantic.alias_generators import to_camel
# 导入 Any 和 Literal 类型注解
from typing import Any, Literal
# 定义 RequestId 类型，既可以是 int 也可以是 str
RequestId = int | str
# 定义 MCP 的基础模型类，支持驼峰命名和按名称填充
class MCPModel(BaseModel):
    # 指定 Pydantic 模型配置：使用驼峰形式命名和按名称填充
   model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
# 定义客户端能力结构体
class ClientCapabilities(MCPModel):
    # 可选字段：客户端实验性能力扩展，键为 str，值为字典
   experimental: dict[str, dict[str, Any]] | None = None
# 定义服务器或客户端的实现信息结构体
class Implementation(MCPModel):
    # 实现的名称
   name: str = ""
    # 实现的版本号
   version: str = ""
    # 可选字段：人类可读的标题
   title: str | None = None
    # 可选字段：实现的描述
   description: str | None = None    
# 定义初始化请求参数结构体
class InitializeRequestParams(MCPModel):
    # 协议版本号
   protocol_version: str = ""
    # 可选字段：客户端能力描述
   capabilities: ClientCapabilities = None
    # 可选字段：客户端实现信息
   client_info: Implementation = None
# 定义初始化请求结构体
class InitializeRequest(MCPModel):
    # 方法名，固定为 "initialize"
   method: Literal["initialize"] = "initialize"
    # 可选字段：参数，类型为 InitializeRequestParams
   params: InitializeRequestParams = None    
# 当前协议的最新版本号
LATEST_PROTOCOL_VERSION = "2024-11-05"
# 定义工具相关能力结构体
class ToolsCapability(MCPModel):
    # 工具列表是否发生变化，可选布尔类型
   list_changed: bool | None = None
# 定义资源相关能力结构体
class ResourcesCapability(MCPModel):
    # 是否支持资源订阅
   subscribe: bool | None = None
    # 资源列表是否变化通知
   list_changed: bool | None = None
# 定义 Prompt 相关能力结构体
+class PromptsCapability(MCPModel):
+   list_changed: bool | None = None
# 定义服务端能力描述结构体
class ServerCapabilities(MCPModel):
    # 可选字段：实验性能力扩展
   experimental: dict[str, dict[str, Any]] | None = None
    # 可选字段：工具能力
   tools: ToolsCapability | None = None
    # 可选字段：资源能力
   resources: ResourcesCapability | None = None
    # 可选字段：Prompt 能力
+  prompts: PromptsCapability | None = None
# 定义初始化响应结构体
class InitializeResult(MCPModel):
    # 协议版本号
   protocol_version: str = ""
    # 可选字段：服务端能力描述
   capabilities: ServerCapabilities = None
    # 可选字段：服务端实现信息
   server_info: Implementation = None
    # 可选字段：初始化说明信息
   instructions: str | None = None
# 定义客户端初始化完成通知结构体
class InitializedNotification(MCPModel):
    # 方法名，固定为 "notifications/initialized"
   method: Literal["notifications/initialized"] = "notifications/initialized"
    # 可选字段：通知参数，可以为字典或 None
   params: dict[str, Any] | None = None    
# 定义 JSONRPC 请求的数据结构
class JSONRPCRequest(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 请求 ID，可以为 int 或 str 类型
    id: RequestId = None
    # 方法名称，字符串类型
    method: str = ""
    # 方法参数，为一个字典或 None
    params: dict[str, Any] | None = None

# 定义 JSONRPC 通知的数据结构（没有 id 字段）
class JSONRPCNotification(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 通知的方法名称
    method: str = ""
    # 通知参数，可以为字典或者 None
    params: dict[str, Any] | None = None

# 定义 JSONRPC 响应的数据结构
class JSONRPCResponse(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 响应的 ID，需要与请求 ID 匹配
    id: RequestId = None
    # 响应结果，可以为字典或 None
    result: dict[str, Any] = None

# 定义错误的数据结构
class ErrorData(BaseModel):
    # 错误码，默认值为 0
    code: int = 0
    # 错误信息，默认值为空字符串
    message: str = ""
    # 附加的错误数据，可以为任意类型或 None
    data: Any = None

# 定义 JSONRPC 错误消息的数据结构
class JSONRPCError(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 错误对应的请求 ID，可以为 None
    id: RequestId | None = None
    # 错误的详细信息，类型为 ErrorData
    error: ErrorData = None
# 定义所有 JSONRPC 消息的联合类型
JSONRPCMessage = JSONRPCRequest | JSONRPCNotification | JSONRPCResponse | JSONRPCError
# 定义 JSONRPC 消息适配器，用于类型自动推断和校验
jsonrpc_message_adapter = TypeAdapter(JSONRPCMessage)
# 定义工具结果中的文本内容块
class TextContent(MCPModel):
    type: Literal["text"] = "text"
    text: str = ""

# 定义工具描述数据结构体
class Tool(MCPModel):
    # 工具名称
   name: str = ""
    # 可选字段：工具描述
   description: str | None = None
    # 工具输入参数的 schema，默认为空字典
   input_schema: dict[str, Any] = Field(default_factory=dict)
    # 可选字段：工具输出 schema
   output_schema: dict[str, Any] | None = None
# 定义获取工具列表请求结构体
class ListToolsRequest(MCPModel):
    # 方法名，固定为 "tools/list"
   method: Literal["tools/list"] = "tools/list"
    # 可选字段：参数，可以为字典或 None
   params: dict[str, Any] | None = None

# 定义获取工具列表响应结构体
class ListToolsResult(MCPModel):
    # 工具列表，默认为空列表
   tools: list[Tool] = []
    # 可选字段：分页游标，可为 None
   next_cursor: str | None = None

# 定义资源元数据结构体
class Resource(MCPModel):
    # 资源 URI
   uri: str = ""
    # 可选：资源名称
   name: str | None = None
    # 可选：人类可读标题
   title: str | None = None
    # 可选：资源描述
   description: str | None = None
    # 可选：MIME 类型
   mime_type: str | None = None

# 定义资源模板结构体
class ResourceTemplate(MCPModel):
    # URI 模板，如 greeting://{name}
   uri_template: str = ""
    # 可选：模板名称
   name: str | None = None
    # 可选：人类可读标题
   title: str | None = None
    # 可选：模板描述
   description: str | None = None
    # 可选：MIME 类型
   mime_type: str | None = None

# 定义 resources/list 请求结构体
class ListResourcesRequest(MCPModel):
   method: Literal["resources/list"] = "resources/list"
   params: dict[str, Any] | None = None

# 定义 resources/list 响应结构体
class ListResourcesResult(MCPModel):
   resources: list[Resource] = []
   next_cursor: str | None = None

# 定义 resources/templates/list 请求结构体
class ListResourceTemplatesRequest(MCPModel):
   method: Literal["resources/templates/list"] = "resources/templates/list"
   params: dict[str, Any] | None = None

# 定义 resources/templates/list 响应结构体
class ListResourceTemplatesResult(MCPModel):
   resource_templates: list[ResourceTemplate] = []
   next_cursor: str | None = None

# 定义 resources/read 请求参数结构体
class ReadResourceRequestParams(MCPModel):
   uri: str = ""

# 定义 resources/read 请求结构体
class ReadResourceRequest(MCPModel):
   method: Literal["resources/read"] = "resources/read"
   params: ReadResourceRequestParams = None

# 定义资源内容基类（文本）
class TextResourceContents(MCPModel):
   uri: str = ""
   mime_type: str | None = None
   text: str = ""

# 定义资源内容基类（二进制）
class BlobResourceContents(MCPModel):
   uri: str = ""
   mime_type: str | None = None
   blob: str = ""  # base64 编码

# 定义 resources/read 响应结构体
class ReadResourceResult(MCPModel):
   contents: list[TextResourceContents | BlobResourceContents] = []

# 定义 Prompt 参数结构体
# 定义代表 Prompt 参数的模型
+class PromptArgument(MCPModel):
    # 参数名称
+   name: str = ""
    # 参数描述，可为 None
+   description: str | None = None
    # 是否为必填参数，可为 None
+   required: bool | None = None

# 定义 Prompt 元数据结构体
# 定义代表 Prompt 元数据的模型
+class Prompt(MCPModel):
    # Prompt 名称
+   name: str = ""
    # Prompt 描述，可为 None
+   description: str | None = None
    # Prompt 参数列表，可为 None
+   arguments: list[PromptArgument] | None = None
    # Prompt 标题，可为 None
+   title: str | None = None

# 定义 prompts/list 请求结构体
# 定义 prompts/list 请求的数据模型
+class ListPromptsRequest(MCPModel):
    # 固定方法字段
+   method: Literal["prompts/list"] = "prompts/list"
    # 请求参数，可为 None
+   params: dict[str, Any] | None = None

# 定义 prompts/list 响应结构体
# 定义 prompts/list 响应的数据模型
+class ListPromptsResult(MCPModel):
    # 返回的 Prompt 列表
+   prompts: list[Prompt] = []
    # 下一页游标，可为 None
+   next_cursor: str | None = None

# 定义 prompts/get 请求参数结构体
# 定义 prompts/get 的参数模型
+class GetPromptRequestParams(MCPModel):
    # Prompt 名称
+   name: str = ""
    # 传递给 Prompt 的参数，可为 None
+   arguments: dict[str, str] | None = None

# 定义 prompts/get 请求结构体
# 定义 prompts/get 请求的数据模型
+class GetPromptRequest(MCPModel):
    # 固定方法字段
+   method: Literal["prompts/get"] = "prompts/get"
    # 请求参数，可为 None
+   params: GetPromptRequestParams | None = None

# 定义 Prompt 消息结构体（role + content）
# 定义表示 Prompt 里的单条消息的数据模型
+class PromptMessage(MCPModel):
    # 消息角色（user 或 assistant）
+   role: Literal["user", "assistant"] = "user"
    # 消息内容，可以是 TextContent 或字典，默认为空文本
+   content: TextContent | dict[str, Any] = Field(default_factory=lambda: TextContent(text=""))

    # 字段校验器：在模型初始化前解析 content 字段
+   @field_validator("content", mode="before")
+   @classmethod
+   def _parse_content(cls, v):
        # 如果是字典且 type 为 "text"，转换为 TextContent 实例
+       if isinstance(v, dict) and v.get("type") == "text":
+           return TextContent(text=v.get("text", ""))
        # 否则原样返回
+       return v

# 定义 prompts/get 响应结构体
# 定义 prompts/get 的响应数据模型
+class GetPromptResult(MCPModel):
    # Prompt 的描述，可为 None
+   description: str | None = None
    # Prompt 返回的消息列表
+   messages: list[PromptMessage] = []


# 定义调用工具请求参数结构体
class CallToolRequestParams(MCPModel):
    # 工具名称
   name: str = ""
    # 可选字段：输入参数，为字典类型或 None
   arguments: dict[str, Any] | None = None        
# 定义调用工具请求结构体
class CallToolRequest(MCPModel):
    # 方法名，固定为 "tools/call"
   method: Literal["tools/call"] = "tools/call"
    # 可选字段：参数，类型为 CallToolRequestParams
   params: CallToolRequestParams = None
# 定义调用工具响应结构体
class CallToolResult(MCPModel):
    # 内容字段，由 TextContent 或字典组成的列表，默认为空列表
    content: list[TextContent | dict[str, Any]] = Field(default_factory=list)
    # 结构化内容字段，可为字典或 None
    structured_content: dict[str, Any] | None = None
    # 错误标志字段，标识是否为错误结果，默认为 False
    is_error: bool = False

    # 针对 content 字段的字段校验器，模型初始化前调用
    @field_validator("content", mode="before")
    @classmethod
    def _parse_content(cls, v):
        # 如果传入的值不是列表，直接返回
        if not isinstance(v, list):
            return v
        # 初始化输出列表
        out = []
        # 遍历每一项
        for item in v:
            # 如果项是字典且类型为 "text"，转换为 TextContent 实例
            if isinstance(item, dict) and item.get("type") == "text":
                out.append(TextContent(text=item.get("text", "")))
            # 否则，原样加入输出列表
            else:
                out.append(item)
        # 返回处理后的列表
        return out       

10. 工作流程 #

10.1 修改概览 #

本次修改在 mcp_lite 中实现了 MCP 的 Prompt（提示词） 能力，使服务端可以注册 Prompt 模板，客户端可以列出并获取这些 Prompt 的渲染结果。

10.1 时序图 #

10.1.1 初始化与列出 Prompt #

10.1.2 获取 Prompt 内容 #

10.1.3 服务端 Prompt 注册与执行流程 #

10.2 各模块说明 #

10.2.1. 类型定义（mcp_lite/types.py） #

新增与 Prompt 相关的类型：

10.2.2. 消息类型（mcp_lite/server/fastmcp/prompts/base.py） #

为 Prompt 返回的消息提供基类和具体类型：

• Message：基类，content 支持 str / TextContent / dict
• UserMessage：role="user"
• AssistantMessage：role="assistant"

Prompt 函数可以返回 str、list[Message] 或 list[dict]，服务端会统一转换为 PromptMessage。

10.2.3. FastMCP 服务端（mcp_lite/server/fastmcp/__init__.py） #

_Prompt 内部类：

• 从函数签名生成参数 schema
• 支持同步/异步函数
• to_prompt() 生成 Prompt 元数据
• run(args) 执行 Prompt 函数

prompt() 装饰器：

@mcp.prompt(name=None, title=None, description=None)
def my_prompt(arg1: str, arg2: str = "default") -> str | list[Message]:
    ...

RPC 处理：

• prompts/list：遍历 _prompts，返回 ListPromptsResult
• prompts/get：根据 name 找到 _Prompt，调用 run(arguments)，将返回值转为 PromptMessage 列表，封装为 GetPromptResult

初始化：若存在已注册 Prompt，在 InitializeResult 中声明 PromptsCapability。

10.2.4. 客户端会话（mcp_lite/client/session.py） #

• list_prompts()：发送 ListPromptsRequest，返回 ListPromptsResult
• get_prompt(name, arguments=None)：发送 GetPromptRequest，返回 GetPromptResult

10.2.5. 示例脚本 #

• prompts_server.py：注册 greet_user（返回 str）和 debug_assistant（返回 list[base.Message]）
• prompts_client.py：同步调用 list_prompts() 和 get_prompt()，并打印消息内容

10.3、数据流小结 #

1. 注册：`@mcp.prompt()将函数包装为_Prompt，存入_prompts`。
2. 列出：prompts/list 遍历 _prompts，返回每个 Prompt 的元数据。
3. 获取：prompts/get 根据 name 找到 _Prompt，用 arguments 调用 run()，将返回值（str / list[Message] / list[dict]）统一转为 PromptMessage 列表，封装为 GetPromptResult 返回。