[原创]fastmcp服务端通信分析

发表于: 2025-9-11 21:39 518

[原创]fastmcp服务端通信分析

Jtian

2025-9-11 21:39

518

项目部署

26cK9s2c8@1M7s2y4Q4x3@1q4Q4x3V1k6Q4x3V1k6Y4K9i4c8Z5N6h3u0Q4x3X3g2U0L8$3#2Q4x3V1k6B7L8r3!0%4K9h3&6Q4x3V1k6X3j5i4y4@1L8h3y4H3
pip install fastmcp
pip install loguru

2e3K9s2c8@1M7s2y4Q4x3@1q4Q4x3V1k6Q4x3V1k6Y4K9i4c8Z5N6h3u0Q4x3X3g2U0L8$3#2Q4x3V1k6o6x3i4m8Z5x3%4u0j5x3e0y4Q4x3V1k6w2j5h3I4A6i4K6u0V1e0f1y4b7

图片描述

Kali-MCP配置

图片描述

注：配的端口转发，访问127.0.0.1:4444相当于访问10.0.2.15:4444，无伤大雅。

初始连通

初始连通时存活探测过了，开关那就是绿色，否则是红的。有一系列的包，其中比较突出的是这个描述功能的包。
图片描述

问题与通信

问题1：Kali-MCP 里都提供了哪些功能？

图片描述
4444端口没有相关流量。

问题2：使用kali mcp计算Kali0911的md5的值

图片描述
MD5值计算正确。

kali_agent执行命令

返回了任务编号，异步执行的

get_task_result查询任务结果
图片描述
返回了任务结果

附件

1.初始连通里描述功能的应答包的json

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "tools": [
      {
        "name": "kali_agent",
        "description": "Execute Kali Linux command asynchronously",
        "inputSchema": {
          "properties": {
            "command": {
              "title": "Command",
              "type": "string"
            },
            "timeout": {
              "anyOf": [
                {
                  "type": "number"
                },
                {
                  "type": "null"
                }
              ],
              "default": 500,
              "title": "Timeout"
            }
          },
          "required": [
            "command"
          ],
          "type": "object"
        },
        "outputSchema": {
          "properties": {
            "result": {
              "title": "Result",
              "type": "string"
            }
          },
          "required": [
            "result"
          ],
          "title": "_WrappedResult",
          "type": "object",
          "x-fastmcp-wrap-result": true
        },
        "_meta": {
          "_fastmcp": {
            "tags": []
          }
        }
      },
      {
        "name": "get_task_result",
        "description": "Get result of a command execution task",
        "inputSchema": {
          "properties": {
            "task_id": {
              "title": "Task Id",
              "type": "string"
            }
          },
          "required": [
            "task_id"
          ],
          "type": "object"
        },
        "outputSchema": {
          "properties": {
            "result": {
              "anyOf": [
                {
                  "additionalProperties": true,
                  "type": "object"
                },
                {
                  "type": "null"
                }
              ],
              "title": "Result"
            }
          },
          "required": [
            "result"
          ],
          "title": "_WrappedResult",
          "type": "object",
          "x-fastmcp-wrap-result": true
        },
        "_meta": {
          "_fastmcp": {
            "tags": []
          }
        }
      }
    ]
  }
}

功能封装代码

图片描述

----20250917补充描述----

问：如何给函数参数添加描述？

答：可以借助annotations来实现。annotations为注释，是字典类型，可以给参数添加备注。
函数体里的'''xxx'''那段描述可能也就增加了可读性，没啥用，因为根本就没有发给大模型，并且实测也是没啥效果。而把参数说明放在annotations里是实测有效的。

@mcp.tool(
    name="pdf_to_text",
    description="将文本型 PDF 转为纯文本（无需 OCR，快速强转）",
    annotations= {
        "file_path":"最好用绝对路径，否则容易找不到文件。"
    },
    enabled=True
)
async def pdf_to_text(file_path: str) -> str:
    logger.info(f"提取任务开始: {file_path}")
    text: str = await asyncio.to_thread(extractor.extract_text_from_path, file_path)
    logger.info("提取任务完成")
    return text

图片描述

{
    "jsonrpc": "2.0",
    "id": 1,
    "result": {
        "tools": [
            {
                "name": "pdf_to_text",
                "description": "将文本型 PDF 转为纯文本（无需 OCR，快速强转）",
                "inputSchema": {
                    "properties": {
                        "file_path": {
                            "title": "File Path",
                            "type": "string"
                        }
                    },
                    "required": [
                        "file_path"
                    ],
                    "type": "object"
                },
                "outputSchema": {
                    "properties": {
                        "result": {
                            "title": "Result",
                            "type": "string"
                        }
                    },
                    "required": [
                        "result"
                    ],
                    "title": "_WrappedResult",
                    "type": "object",
                    "x-fastmcp-wrap-result": true
                },
                "annotations": {
                    "file_path": "最好用绝对路径，否则容易找不到文件。"
                },
                "_meta": {
                    "_fastmcp": {
                        "tags": []
                    }
                }
            }
        ]
    }
}

tool装饰器函数原型

tool装饰器函数原型在Lib/site-packages/fastmcp/server/server.py

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

def tool(
    self,
    name_or_fn: str | AnyFunction | None = None,
    *,
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    tags: set[str] | None = None,
    output_schema: dict[str, Any] | None | NotSetT = NotSet,
    annotations: ToolAnnotations | dict[str, Any] | None = None,
    exclude_args: list[str] | None = None,
    meta: dict[str, Any] | None = None,
    enabled: bool | None = None,
) -> Callable[[AnyFunction], FunctionTool] | FunctionTool:
    """Decorator to register a tool.
 
    Tools can optionally request a Context object by adding a parameter with the
    Context type annotation. The context provides access to MCP capabilities like
    logging, progress reporting, and resource access.
 
    This decorator supports multiple calling patterns:
    - @server.tool (without parentheses)
    - @server.tool (with empty parentheses)
    - @server.tool("custom_name") (with name as first argument)
    - @server.tool(name="custom_name") (with name as keyword argument)
    - server.tool(function, name="custom_name") (direct function call)
 
    Args:
        name_or_fn: Either a function (when used as @tool), a string name, or None
        name: Optional name for the tool (keyword-only, alternative to name_or_fn)
        description: Optional description of what the tool does
        tags: Optional set of tags for categorizing the tool
        output_schema: Optional JSON schema for the tool's output
        annotations: Optional annotations about the tool's behavior
        exclude_args: Optional list of argument names to exclude from the tool schema
        meta: Optional meta information about the tool
        enabled: Optional boolean to enable or disable the tool
 
    Examples:
        Register a tool with a custom name:
        ```python
        @server.tool
        def my_tool(x: int) -> str:
            return str(x)
 
        # Register a tool with a custom name
        @server.tool
        def my_tool(x: int) -> str:
            return str(x)
 
        @server.tool("custom_name")
        def my_tool(x: int) -> str:
            return str(x)
 
        @server.tool(name="custom_name")
        def my_tool(x: int) -> str:
            return str(x)
 
        # Direct function call
        server.tool(my_function, name="custom_name")
        ```
    """
    if isinstance(annotations, dict):
        annotations = ToolAnnotations(**annotations)
         
    if isinstance(name_or_fn, classmethod):
        raise ValueError(
            inspect.cleandoc(
                """
                To decorate a classmethod, first define the method and then call
                tool() directly on the method instead of using it as a
                decorator. See cfbK9s2c8@1M7s2y4Q4x3@1q4Q4x3V1k6Q4x3V1k6Y4L8$3k6S2M7%4c8E0j5%4m8Q4x3X3g2U0L8$3#2Q4x3V1k6H3j5i4c8@1k6i4u0F1M7#2)9J5c8X3c8W2j5$3!0J5j5i4c8A6L8X3N6Q4x3X3c8E0k6i4c8Z5L8$3c8K6
                for examples and more information.
                """
            )
        )
 
    # Determine the actual name and function based on the calling pattern
    if inspect.isroutine(name_or_fn):
        # Case 1: @tool (without parens) - function passed directly
        # Case 2: direct call like tool(fn, name="something")
        fn = name_or_fn
        tool_name = name  # Use keyword name if provided, otherwise None
 
        # Register the tool immediately and return the tool object
        tool = Tool.from_function(
            fn,
            name=tool_name,
            title=title,
            description=description,
            tags=tags,
            output_schema=output_schema,
            annotations=cast(ToolAnnotations | None, annotations),
            exclude_args=exclude_args,
            meta=meta,
            serializer=self._tool_serializer,
            enabled=enabled,
        )
        self.add_tool(tool)
        return tool
 
    elif isinstance(name_or_fn, str):
        # Case 3: @tool("custom_name") - name passed as first argument
        if name is not None:
            raise TypeError(
                "Cannot specify both a name as first argument and as keyword argument. "
                f"Use either @tool('{name_or_fn}') or @tool(name='{name}'), not both."
            )
        tool_name = name_or_fn
    elif name_or_fn is None:
        # Case 4: @tool or @tool(name="something") - use keyword name
        tool_name = name
    else:
        raise TypeError(
            f"First argument to @tool must be a function, string, or None, got {type(name_or_fn)}"
        )
 
    # Return partial for cases where we need to wait for the function
    return partial(
        self.tool,
        name=tool_name,
        title=title,
        description=description,
        tags=tags,
        output_schema=output_schema,
        annotations=annotations,
        exclude_args=exclude_args,
        meta=meta,
        enabled=enabled,
    )