Python-MCPServer开发

Python-MCPServer开发

使用FastMCP开发【SSE模式的MCPServer】，熟悉【McpServer编码过程】+【McpServer调试方法】

1-核心知识点

  • 1-熟悉【SSE模式的MCPServer】开发
  • 2-熟悉【stdio模式的MCPServer】开发
  • 3-熟悉【启动MCPServer】的三种方式
    • 3.1-直接启动:python mcp_server.py
    • 3.2-代码中uv使用：subprocess.Popen([“uv”, “run”, server_file])
    • 3.3-使用mcpInspector网页：mcp dev mcp_server.py
  • 4-熟悉【Agent增强外部能力】的两种方式
    • 4.1-调用tool:Agent(name=“Assistant”, instructions=“用中文回复”, tools=[get_weather])
    • 4.2-调用tool:Agent(name=“Assistant”, instructions=“用中文回复”, mcp_servers=[mcp_server])

2-思路整理

1-McpServer须知

  • 1-MCPServer已经可以对外提供服务了，这个不同于传统的API，虽然也运行在特殊端口了，但是不能通用postman进行调用

  • 2-那有人质疑这个说法了，为什么使用mcpInspector网页调试就可以连通？

    • 可以理解为：mcpInspector联通的是Agent，只是查看Agent上有哪些Resources/Prompts/Tools…并且可以通过Agent和McpServer进行交互，而不是Rest请求

  • 3-MCPServer_SSE和MCPServer_stdio如何进行快速区分：

    • 1-stdio-适用于同一台机器上（进程间通讯，通讯不经过HTTP协议，超级快）->要求：你要有可运行文件【Python文件】或者【Java的Jar包】
    • 2-SSE适用于所有环境，通过IP进行HTTP通讯->基本没有要求，无脑就全部使用这种

  • 4-再次澄清：MCPServer_SSE虽然可以部署，有大网IP或者域名，但是依然不能通用postman进行调用

    • post调用的接口->要求是@Controller标记或者FastAPI开发的API接口
    • agent调用的接口->要求是@mcp.tool标记（可能对外就不是API接口）
    • 总结：可以理解@mcp.tool标记后是一种特殊的API，只能被Agent调用

  • 5-SSE和stdio在代码上有什么区别？

    • 1-McpServer的编写上只有一个区别【mcp.run(transport=“sse/stdio”)】
    • 2-Agent在连接的时候使用【McpServerstdio】和【McpServerSSE】进行McpServer构建
    • 3-【mcp.run(transport=“sse”)】后，服务【可以被任何远程】的Agent进行访问（此时先不谈鉴权的问题）
    • 4-【mcp.run(transport=“stdio”)】后，服务【只可以被本地进程】的Agent进行访问（此时先不谈鉴权的问题）

2-McpServer核心思路

1-【mcp.run(transport=“sse”)】后，服务【可以被任何远程】的Agent进行访问（此时先不谈鉴权的问题）

2-【mcp.run(transport=“stdio”)】后，服务【只可以被本地进程】的Agent进行访问（此时先不谈鉴权的问题）

  • 1-编写传统的Service业务代码
  • 2-在Service业务代码头上添加@tool装饰器，即可实现FastMCP的Tool功能
  • 3-在Service业务代码头上添加@mcp.tool()装饰器，即可实现FastMCP的McpServer功能
  • 4-主程序指定运行方法-stdio进程启动（此时MCPServer已经可以对外提供服务了，这个不同于传统的API，虽然也运行在特殊端口了，但是不能通用postman进行调用）
    • 1-直接启动:python mcp_server.py
    • 2-代码中uv使用：subprocess.Popen([“uv”, “run”, server_file])
    • 3-使用mcpInspector网页：mcp dev mcp_server.py
  • 5-Agent调试McpServer
    • 1-指定LLM
    • 2-指定Agent(McpServerList+角色定位)
    • 3-Runner.run(Agent+LLM+问题)交互

3-McpServer和McpClient关系

  • 1-McpClient只是一个概念，在使用Python进行开发的时候根本就没有任何关于McpClient的类
  • 2-McpClient就是McpServer连接过程的过程

4-Agent+LLM+McpServers

  • 1-LLM和Agent最初是独立创建的，两者之间没有关系

    • LLM指明用哪家大模型
    • Agent指明用哪些McpServerList

  • 2-LLM和Agent是通过Runner.run(Agent+LLM+问题)进行关联

  • 3-Agent+LLM如何进行交互（要不要进行最后的总结）是通用（result.output获取的时候进行逻辑编排的）

    • 如果我们自己做MCP编排，可能就是在这个地方进行核心逻辑编写

5-MCPServer核心代码

  • 1-在Service业务代码头上添加@tool装饰器，即可实现FastMCP的Tool功能
# 假设 mcp 已经正确导入
try:from mcp import tool
except ImportError:# 如果 mcp 未找到，模拟一个 tool 装饰器def tool(func):return func# 在 Service 业务代码头上添加 @tool 装饰器
@tool
async def get_city_list(self) -> list:"""获取所有的城市信息。返回:str: 所有的城市信息列表"""logging.info(f"获取所有的城市信息")city_list = []for city in self.CITY_WEATHER_DATA:city_list.append(city)return city_list
  • 2-在Service业务代码头上添加@mcp.tool()装饰器，即可实现FastMCP的McpServer功能
from mcp.server.fastmcp import FastMCPfrom city_01_service import CityDataServer# 1-初始化 MCP 服务器
mcp = FastMCP("CityDataServer")# 2-初始化城市信息服务器(业务代码+@tool装饰器)
city_server = CityDataServer()# 3-在 Service 业务代码头上添加@mcp.tool()装饰器
@mcp.tool()
# 获取所有城市列表
async def get_city_list():"""获取所有城市列表。返回:str: 所有城市列表"""city_list = await city_server.get_city_list()return city_list# 4-主程序指定运行方法-stdio/sse进程启动
if __name__ == "__main__":mcp.run(transport='stdio/sse')

3-参考网址

  • MCP官网的开发样例：https://github.com/openai/openai-agents-python/blob/main/examples/mcp/sse_example/main.py
  • 个人代码实现仓库：https://gitee.com/enzoism/python_mcp_client_server

4-上手实操

1-空工程初始化环境

mkdir my_project
cd my_project
python -m venv .venv

2-激活环境

# Windows
source .venv/Scripts/activate# Mac
source .venv/bin/activate

3-添加依赖

对应的依赖是在激活的环境中

# uv用于后续MCP Inspector的连接
pip install uv httpx mcp

4-ToolFunction核心代码

import asyncio
import osfrom agents import (Agent,RunConfig,Runner,function_tool,set_tracing_disabled,
)
from dotenv import load_dotenvfrom model_providers.deepseek import DeepSeekModelProvider# 1-环境变量加载相关
load_dotenv()
BASE_URL = os.getenv("BASE_URL") or ""
API_KEY = os.getenv("API_KEY") or ""
MODEL_NAME = os.getenv("MODEL_NAME") or ""
if not BASE_URL or not API_KEY or not MODEL_NAME:raise ValueError("请通过环境变量或代码设置EXAMPLE_BASE_URL、EXAMPLE_API_KEY、EXAMPLE_MODEL_NAME。")# 2-跳过大模型的链路追踪
set_tracing_disabled(disabled=True)"""
本例使用自定义提供程序调用Runner.run()的部分，并直接调用OpenAI进行其他操作。
步骤：
1. 【实例化LLM】ModelProvider对象-并构建RunConfig
2. 【实例化Agent】创建一个Agent。
3. 在调用Runner.run()结合【LLM】+【Agent】进行问答
注意，在本例中，我们假设您没有从platform.openai.com获取API密钥，因此禁用了跟踪。
如果您有API密钥，您可以选择设置`OPENAI_API_KEY`环境变量或调用set_tracing_export_api_key()来设置跟踪特定的密钥。
"""# 3-定义一个工具函数
@function_tool
def init_weather_tool_function(city: str):print(f"[debug] getting weather for {city}")return f"The weather in {city} is sunny."async def run_mcp_tool_function():# 1-【实例化LLM】ModelProvider对象-并构建RunConfigrun_config = RunConfig(model_provider=DeepSeekModelProvider(BASE_URL, API_KEY, MODEL_NAME))# 2-【实例化Agent】创建一个Agentagent = Agent(name="Assistant",instructions="使用工具回答大模型的问题",tools=[init_weather_tool_function])# 3.1-调用工具回答问题message = "杭州的天气怎么样？"print(f"\n\n【大模型请求案例】-> {message}")result = await Runner.run(starting_agent=agent, input=message, run_config=run_config)print(result.final_output)# 3.2-获取Agent对话的结果-没有配置RunConfig使用的是OpenAI的默认模型# result = await Runner.run(#     agent,#     "给我讲一个笑话吧!",# )# print(result.final_output)if __name__ == "__main__":asyncio.run(run_mcp_tool_function())

5-McpServer核心代码

1-【mcp.run(transport=“sse”)】后，服务【可以被任何远程】的Agent进行访问（此时先不谈鉴权的问题）

2-【mcp.run(transport=“stdio”)】后，服务【只可以被本地进程】的Agent进行访问（此时先不谈鉴权的问题）

import random
import randomfrom mcp.server.fastmcp import FastMCP# Create server
mcp = FastMCP("Echo Server")# 1-MCP工具1-数字加和
@mcp.tool()
def add(a: int, b: int) -> int:"""Add two numbers"""print(f"[MCP工具1-数字加和] add({a}, {b})")return a + b# 2-MCP工具2-随机选定的字符
@mcp.tool()
def get_secret_word() -> str:"""获取随机单词"""print("[MCP工具2-随机选定的字符] get_secret_word()")return random.choice(["apple", "banana", "cherry"])if __name__ == "__main__":mcp.run(transport="sse/stdio")

6-Agent调试McpServer

hitokoto_02_api.py：API直接调用service对外暴露rest请求

import asyncio
import osfrom agents import Agent, Runner, RunConfig, set_tracing_disabled
from agents.mcp import MCPServer, MCPServerStdio
from agents.model_settings import ModelSettings
from dotenv import load_dotenvfrom model_providers.deepseek import DeepSeekModelProvider# 1-环境变量加载相关
load_dotenv()
BASE_URL = os.getenv("BASE_URL") or ""
API_KEY = os.getenv("API_KEY") or ""
MODEL_NAME = os.getenv("MODEL_NAME") or ""
if not BASE_URL or not API_KEY or not MODEL_NAME:raise ValueError("请通过环境变量或代码设置EXAMPLE_BASE_URL、EXAMPLE_API_KEY、EXAMPLE_MODEL_NAME。")# 2-跳过大模型的链路追踪
set_tracing_disabled(disabled=True)"""
本例使用自定义提供程序调用Runner.run()的部分，并直接调用OpenAI进行其他操作。
步骤：
1. 【实例化LLM】ModelProvider对象-并构建RunConfig
2. 【实例化Agent】创建一个Agent。
3. 在调用Runner.run()结合【LLM】+【Agent】进行问答
- 1）直接和大模型对话
- 2）调用MCPServer_SSE模式-[MCP工具1-数字加和]
- 3）调用MCPServer_SSE模式-[MCP工具2-随机选定的字符]
"""async def run_mcp_server(mcp_server: MCPServer):# 1-【实例化LLM】ModelProvider对象-并构建RunConfigrun_config = RunConfig(model_provider=DeepSeekModelProvider(BASE_URL, API_KEY, MODEL_NAME))# 2-【实例化Agent】创建一个Agentagent = Agent(name="Assistant",instructions="使用工具回答大模型的问题",mcp_servers=[mcp_server],model_settings=ModelSettings(tool_choice="required"),)# 3.1-直接和大模型对话message = "给我讲一个笑话吧！"print(f"\n\n【大模型请求案例】-> {message}")result = await Runner.run(starting_agent=agent, input=message, run_config=run_config)print(result.final_output)# 3.2-调用MCPServer_SSE模式-[MCP工具1-数字加和]message = "What's the weather in Tokyo?"print(f"\n\n【大模型请求案例】-> {message}")result = await Runner.run(starting_agent=agent, input=message, run_config=run_config)print(result.final_output)# 3.3-调用MCPServer_SSE模式-[MCP工具2-随机选定的字符]message = "What's the secret word?"print(f"\n\n【大模型请求案例】-> {message}")result = await Runner.run(starting_agent=agent, input=message, run_config=run_config)print(result.final_output)async def init_mcp_server() -> MCPServerStdio:# 1-创建 MCP 服务器连接实例，但不立即运行(python mcp_server_xx.py)this_dir = os.path.dirname(os.path.abspath(__file__))python_exec_path = os.path.join(this_dir, ".venv/Scripts/python.exe")mcp_server_stdio_file = os.path.join(this_dir, "mcp05_stdio.py")weather_server = MCPServerStdio(name="weather",params={"command": python_exec_path,"args": [mcp_server_stdio_file],"env": {},},# 缓存工具列表以减少延迟，不需要每次连接时重新获取工具列表cache_tools_list=True)# 2-手动连接到MCP服务器print("正在连接到MCP服务器...")await weather_server.connect()print("MCP服务器连接成功！")# 3-等待服务器连接成功并获取MCP服务可用工具列表tools = await weather_server.list_tools()print("\n可用工具列表: ")for tool in tools:print(f" - {tool.name}: {tool.description}")return weather_serverif __name__ == "__main__":# 1-获取McpServerserver = asyncio.run(init_mcp_server())# 2-运行McpServerasyncio.run(run_mcp_server(server))

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如若转载，请注明出处：http://www.pswp.cn/bicheng/78960.shtml
繁体地址，请注明出处：http://hk.pswp.cn/bicheng/78960.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

高级项目管理

高级项目管理

在信息系统项目管理工作中，组织管理者和项目管理者，有时还会面临多项目的管理，或组织级的项目管理、项目的量化管理等课题。 其中，项目集管理、项目组合管理和组织级项目管理，为多项目管理和组织级管理提供有效指导&a…
阅读更多...
tarjan缩点+强联通分量

tarjan缩点+强联通分量

【模板】缩点https://www.luogu.com.cn/problem/P3387 首先我们要理解这道题为什么要用缩点 题目说的是有向图，如果无环的话就可以用DP来解决了 由于可以走重复的点，所以一个环上的点可以看成是一个点，它的点权就等于该环上所有点的点权之…
阅读更多...
OSCP：获取全交互式 Windows 反向 Shell

OSCP：获取全交互式 Windows 反向 Shell

简介 在本文中，我们将探讨获取完全交互式 Windows 反向 Shell 的各种方法，从利用内置工具到采用先进技术以获得更好的控制和功能。 通过 Invoke-ConPtyShell 我获取完全交互式 Windows 反向 Shell 的首选方法是通过 Invoke-ConPtyShell 脚本。当 Wind…
阅读更多...
免费超好用的电脑操控局域网内的手机（多台，无线）

免费超好用的电脑操控局域网内的手机（多台，无线）

使用 第一步 解压QtScrcpy压缩包，并运行QtScrcpy.exe 第二步 2.1 手机开启开发者模式（设置>关于本机>版本信息>连点10下“版本号”） 2.2 开启 USB调试 和 无线调试（设置>开发者选项> USB调试 无线调试&#xf…
阅读更多...
Go语言内存管理

Go语言内存管理

本章节，就来学习一下go语言的内存模型，看一下内存的分配，存储都是如何实现的，与此同时，在正式开始今天的主题之前，首先先来学习操作系统基于这一方面的内容，来看看是如何管理内存的吧 本章及节…
阅读更多...
【docker】启动临时MongoDB容器、挂载数据卷运行数据库服务，并通过备份文件恢复MongoDB数据库备份数据

【docker】启动临时MongoDB容器、挂载数据卷运行数据库服务，并通过备份文件恢复MongoDB数据库备份数据

‌启动临时 MongoDB 容器、挂载数据卷运行数据库服务，并通过备份文件恢复数据 1.命令分解与功能说明1.1.启动一个临时 MongoDB 容器‌，并进入交互式终端（1）执行命令（2）实现功能‌（3）…
阅读更多...
【最新 MCP 战神手册 08】工具使用详解：实现 AI 行动

【最新 MCP 战神手册 08】工具使用详解：实现 AI 行动

文章目录 1. 开始啦！2. 第一部分：设计高效且安全的工具3. 第二部分：定义工具蓝图——参数、输出与约束条件4. 第三部分：弥合差距：LLM 兼容性（函数调用）5. 第四部分：实施与测试的最佳实践1. 开始啦！ 在前几章中，我们将工具介绍为 AI 模型在 MCP 客户端引导下向 MCP 服…
阅读更多...
介绍 IntelliJ IDEA 快捷键操作

介绍 IntelliJ IDEA 快捷键操作

IntelliJ IDEA 快捷键操作 1. 编辑与导航2. 查找与替换3. 调试与运行4. 导航与视图5. 重构与生成6. 高级快捷键（提高效率）注意事项 IntelliJ IDEA 是一款功能强大的集成开发环境，掌握其常用快捷键可以显著提升开发效率。但是有些小伙伴并不清…
阅读更多...
Javascript 中作用域的理解?

Javascript 中作用域的理解?

一、作用域的类型 1. 全局作用域（公司大门外） 范围：整个 JavaScript 文件变量：像贴在公告栏上的信息，所有人可见例子：const companyName "阿里"; // 全局变量，任何地方都能访问 fu…
阅读更多...
Leetcode刷题记录22——滑动窗口最大值

Leetcode刷题记录22——滑动窗口最大值

题源：https://leetcode.cn/problems/sliding-window-maximum/description/?envTypestudy-plan-v2&envIdtop-100-liked 题目描述： 思路一： 暴力遍历法，通过一个长度为k的滑动窗口遍历nums，将其中最大的数依次记…
阅读更多...
Apache Flink的架构设计与运行流程说明

Apache Flink的架构设计与运行流程说明

在大数据领域，实时计算的重要性随着业务需求的爆发式增长愈发凸显。从电商的实时销量监控到金融的高频交易风控，从物联网设备的实时告警到社交平台的热点追踪，企业对“秒级甚至毫秒级”数据处理能力的需求已成为刚需。在众多实时计算框架中&a…
阅读更多...
经典算法 最长单调递增子序列

经典算法 最长单调递增子序列

最长单调递增子序列 问题描述 找出由n个数组成的序列的最长单调递增子序列。 示例输入 9 2 1 5 3 6 4 8 9 7示例输出 5示例输入 6 5 6 7 1 2 8示例输出 4c代码(动态规划 O(n^2)) #include<bits/stdc.h>using namespace std;int main() {int n, ans 0;cin >&g…
阅读更多...
【语法】C++继承中遇到的问题及解决方法

【语法】C++继承中遇到的问题及解决方法

目录 1.子类构造函数中初始化父类成员 2.子类显式调用父类的析构函数 第一种说法&#xff1a;重定义 反驳&#xff1a; 第二种说法&#xff1a;operator~ 3.因编译器版本过低而出现错误 贴主在学习C的继承时&#xff0c;遇到了很多问题&#xff0c;觉得很变态&#xff0c…
阅读更多...
前缀和 后缀和 --- 寻找数组的中心下标

前缀和 后缀和 --- 寻找数组的中心下标

题目链接 寻找数组的中心下标 给你一个整数数组 nums &#xff0c;请计算数组的 中心下标 。 数组 中心下标 是数组的一个下标&#xff0c;其左侧所有元素相加的和等于右侧所有元素相加的和。 如果中心下标位于数组最左端&#xff0c;那么左侧数之和视为 0 &#xff0c;因为…
阅读更多...
NVIDIA --- 端到端自动驾驶

NVIDIA --- 端到端自动驾驶

前言 参加了NVIDIA 高级辅助驾驶开发者实验室的活动&#xff0c;本次活动基于 NVIDIA 汽车行业的端到端解决方案——DRIVE AGX™ 平台&#xff0c;实现高级别智能和安全性的软硬件开发工具和 AV 基础设施。并且NVIDIA自动驾驶实验室推出了一系列自动驾驶算法最新的前沿研究视频…
阅读更多...
SQL实战：03之SQL中的递归查询

SQL实战：03之SQL中的递归查询

文章目录 概述SQL 中的递归实现题目一:分析组织层级题解题目二:树节点求解题解步骤一&#xff1a;通过递归查询出每个节点的上级节点和下级节点分布步骤二&#xff1a;分组统计 概述 最近刷题时遇到了一道需要根据组织层级来统计各个层级的一些数据&#xff0c;当时碰到时的第…
阅读更多...
MySQL 语法与基础完全指南

MySQL 语法与基础完全指南

MySQL 是最流行的开源关系型数据库管理系统之一&#xff0c;广泛应用于 Web 应用程序开发。本文将全面介绍 MySQL 的基础知识和完整语法结构。 一、MySQL 基础概念 1. 数据库基本术语 数据库(Database): 存储数据的集合 表(Table): 数据以表格形式组织 列(Column): 表中的一…
阅读更多...
【Sqlalchemy Model转换成Pydantic Model示例】

【Sqlalchemy Model转换成Pydantic Model示例】

【Sqlalchemy Model转换成Pydantic Model示例】 由于Sqlalchemy和Pydantic的模型字段类型可能有差异, 所以需要一个通用的装换类 def sqlalchemy_to_pydantic_v2(sqlalchemy_model, pydantic_model):"""通用函数&#xff0c;将 SQLAlchemy 模型实例转换为 Pyd…
阅读更多...
2025年欧洲西南部大停电

2025年欧洲西南部大停电

2025年4月28日&#xff0c;欧洲西南部出现大规模停电&#xff0c;西班牙、葡萄牙和法国南部均受到影响。有报道指出停电可能与 欧洲电网出现问题有关&#xff0c;但最终原因尚未确定。由于停电&#xff0c;上述地区的交通和通信服务均受到严重影响&#xff0c;交通信号灯停止工…
阅读更多...
Java EE初阶——计算机是如何工作的

Java EE初阶——计算机是如何工作的

1. cpu 冯诺依曼体系&#xff08;Von Neumann Architecture&#xff09; • CPU 中央处理器: 进⾏算术运算和逻辑判断. • 存储器: 分为外存和内存, ⽤于存储数据(使⽤⼆进制⽅式存储) • 输⼊设备: ⽤⼾给计算机发号施令的设备. • 输出设备: 计算机个⽤⼾汇报结果的设…
阅读更多...
最新文章

Copyright @ 2022~2023