一组独立的Python脚本,实现了各种实用功能的模型上下文协议(MCP)服务器。每个服务器提供专门的工具,可以被支持MCP协议的AI助手或其他应用程序使用。
模型上下文协议(MCP)是一种标准化的方式,用于AI助手与外部工具和服务进行交互。它允许AI模型通过调用由MCP服务器提供的专用函数来扩展其功能。通信通过标准输入/输出(stdio)使用JSON消息进行。
每个MCP服务器都设计为使用Python环境管理器如uv运行。
ytdlp)一个使用yt-dlp从YouTube视频中提取信息的服务器。
工具:
MCP服务器配置:
"mcpServers": {
"ytdlp": {
"name": "youtube", // 客户端可选友好名称
"command": "uv",
"args": [
"run",
"--directory", "<path/to/repo>/useful-mcps/ytdlp", // 包含pyproject.toml的MCP目录路径
"--", // 脚本参数前的分隔符,如果有
"ytdlp_mcp" // 匹配pyproject.toml [project.scripts] 中定义的脚本名称
]
// 使用 --directory 时不需要 'cwd'
}
}
docx_replace)一个用于操作Word文档的服务器,包括模板处理和PDF转换。
工具:
MCP服务器配置:
"mcpServers": {
"docx_replace": {
"name": "docx", // 可选友好名称
"command": "uv",
"args": [
"run",
"--directory", "<path/to/repo>/useful-mcps/docx_replace", // MCP目录路径
"--",
"docx_replace_mcp" // 匹配pyproject.toml中定义的脚本名称
]
}
}
plantuml)一个使用PlantUML服务器渲染PlantUML图表的服务器(通常通过Docker运行)。
工具:
MCP服务器配置:
"mcpServers": {
"plantuml": {
"name": "plantuml", // 可选友好名称
"command": "uv",
"args": [
"run",
"--directory", "<path/to/repo>/useful-mcps/plantuml", // MCP目录路径
"--",
"plantuml_server" // 匹配pyproject.toml中定义的脚本名称
]
}
}
(注意:需要一个可访问的运行中的PlantUML服务器,可能通过Docker管理)
mermaid)一个使用mermaidchart.com API渲染Mermaid图表的服务器。
工具:
MCP服务器配置:
"mcpServers": {
"mermaid": {
"name": "mermaid", // 可选友好名称
"command": "uv",
"args": [
"run",
"--directory", "<path/to/repo>/useful-mcps/mermaid", // MCP目录路径
"--",
"mermaid_mcp" // 匹配pyproject.toml中定义的脚本名称
],
"env": { // MCP所需的环境变量
"MERMAID_CHART_ACCESS_TOKEN": "YOUR_API_TOKEN_HERE"
}
}
}
(注意:需要设置Mermaid Chart API访问令牌作为环境变量)
rss2md)一个将RSS源内容转换为Markdown格式的服务器,具有日期过滤功能。
工具:
MCP服务器配置:
"mcpServers": {
"mermaid": {
"name": "rss2md", // 可选友好名称
"command": "uv",
"args": [
"run",
"--directory", "<path/to/repo>/useful-mcps/rss2md", // MCP目录路径
"--",
"rss2md_mcp" // 匹配pyproject.toml中定义的脚本名称
],
"env": { // MCP所需的环境变量
}
}
}
克隆仓库:
git clone https://github.com/daltonnyx/useful-mcps.git # 如果实际仓库URL不同,请替换
cd useful-mcps
安装uv:如果没有安装uv,请安装:
pip install uv
# 或者按照 https://github.com/astral-sh/uv 的说明进行
依赖项:依赖项通过pyproject.toml按MCP管理。uv run通常会在首次使用--directory运行MCP时自动在虚拟环境中安装它们。
建议使用uv run --directory <path>指向特定MCP的目录来运行每个MCP服务器。uv根据该目录下的pyproject.toml处理虚拟环境和依赖项。
示例(从根useful-mcps目录):
# 运行YouTube MCP
uv run --directory ./ytdlp ytdlp_mcp
# 运行Mermaid MCP(确保环境变量中的令牌已设置)
uv run --directory ./mermaid mermaid_mcp
或者,配置您的MCP客户端(如上面的示例JSON配置)直接执行uv run --directory ...命令。
配置您的MCP客户端应用程序,使用“MCP服务器配置”示例中显示的command和args结构启动所需的服务器。确保command指向您的uv可执行文件,而args正确指定--directory以及MCP文件夹的路径和要运行的脚本名称。使用env属性传递必要的环境变量(如API令牌)。
这些展示了您将发送给相应MCP服务器的call_tool函数的arguments示例。
{
"url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
}
{
"url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
"language": "en",
"chapters": [
{
"title": "Introduction",
"start_time": "00:00:00",
"end_time": "00:01:30"
}
]
}
{
"template_file": "/path/to/template.docx",
"replacements": {
"name": "John Doe",
"date": "2023-05-15"
},
"blocks": {
"optional_section": true,
"alternative_section": false
},
"output_filename": "/path/to/output.docx"
}
(注意:template_file和docx_file也可以接受base64编码字符串而不是路径)
{
"template_file": "/path/to/template.docx"
}
{
"docx_file": "/path/to/document.docx",
"pdf_output": "/path/to/output.pdf"
}
{
"input": "participant User\nUser -> Server: Request\nServer --> User: Response",
"output_path": "/path/to/save/diagram.png"
}
(注意:input也可以是一个.puml文件的路径)
{
"mermaid_code": "graph TD;\n A-->B;\n A-->C;\n B-->D;\n C-->D;",
"output_path": "/path/to/save/mermaid.png",
"theme": "default" // 可选,例如:"default", "dark", "neutral", "forest"
}
my_new_mcp)。pyproject.toml:定义项目元数据、依赖项和脚本入口点(例如,[project.scripts]部分映射my_new_mcp = "my_new_mcp:main")。pyrightconfig.json:(可选)用于类型检查。my_new_mcp.py):使用mcp库实现MCP逻辑(参见下面的模板)。serve,list_tools,call_tool)。基本模板(my_new_mcp.py):
import json
import logging
import asyncio
from typing import List, Dict, Any, Optional
# 假设mcp库已安装或可用
# from mcp import Server, Tool, TextContent, stdio_server
# 如果mcp库结构不同,使用占位符导入
from typing import Protocol # 使用Protocol作为占位符
# 如果mcp库不能直接导入,使用占位符定义
class Tool(Protocol):
name: str
description: str
inputSchema: dict
class TextContent(Protocol):
type: str
text: str
class Server:
def __init__(self, name: str): pass
def list_tools(self): pass # 装饰器
def call_tool(self): pass # 装饰器
def create_initialization_options(self): pass
async def run(self, read_stream, write_stream, options): pass
# 占位符上下文管理器
class stdio_server:
async def __aenter__(self): return (None, None) # 虚拟流
async def __aexit__(self, exc_type, exc, tb): pass
# Pydantic常用于模式定义
# from pydantic import BaseModel
# class MyInput(BaseModel):
# param1: str
# param2: int
class MyInputSchema: # 如果不使用Pydantic,使用占位符
@staticmethod
def model_json_schema():
return {"type": "object", "properties": {"param1": {"type": "string"}, "param2": {"type": "integer"}}, "required": ["param1", "param2"]}
class MyTools:
TOOL_NAME = "my.tool"
class MyService:
def __init__(self):
# 初始化资源如果需要
pass
def my_function(self, param1: str, param2: int) -> dict:
# 实现您的工具功能
logging.info(f"正在运行my_function,{param1=}, {param2=}")
# 替换为实际逻辑
result_content = f"结果:处理了{param1}和{param2}"
return {"content": result_content}
async def serve() -> None:
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
server = Server("mcp-my-service")
service = MyService()
@server.list_tools()
async def list_tools() -> list[Tool]:
logging.info("list_tools被调用")
return [
Tool(
name=MyTools.TOOL_NAME,
description="我的工具描述",
# 使用Pydantic的模式或手动定义
inputSchema=MyInputSchema.model_json_schema(),
),
]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
logging.info(f"call_tool被调用,{name=}, {arguments=}")
try:
if name == MyTools.TOOL_NAME:
# 如果不使用Pydantic,这里添加验证
param1 = arguments.get("param1")
param2 = arguments.get("param2")
if param1 is None or param2 is None:
raise ValueError("缺少必需的参数")
result = service.my_function(param1, int(param2)) # 确保需要时进行类型转换
logging.info(f"工具执行成功:{result=}")
return [TextContent(type="text", text=json.dumps(result))] # 返回JSON字符串
else:
logging.warning(f"请求未知工具:{name}")
raise ValueError(f"未知工具:{name}")
except Exception as e:
logging.error(f"执行工具{name}时出错:{e}", exc_info=True)
# 将错误作为JSON返回
error_payload = json.dumps({"error": str(e)})
return [TextContent(type="text", text=error_payload)]
options = server.create_initialization_options()
logging.info("正在启动MCP服务器...")
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, options)
logging.info("MCP服务器已停止。")
def main():
# 在pyproject.toml `[project.scripts]`中定义的入口点
try:
asyncio.run(serve())
except KeyboardInterrupt:
logging.info("用户中断服务器。")
if __name__ == "__main__":
# 允许直接通过`python my_new_mcp.py`运行以调试
main()
从根目录运行测试:
pytest tests/
(确保安装了测试依赖项,可以通过uv pip install pytest或在其中一个pyproject.toml文件中将pytest添加到开发依赖项来实现)
欢迎贡献!请随意提交拉取请求。