🚀 MCP 服务器模板
MCP 服务器模板为开发提供了基础架构,涵盖项目结构、依赖管理、服务器启动等功能,能帮助开发者快速搭建 MCP 服务器。
🚀 快速开始
1. 安装依赖
pip install -r requirements.txt
2. 启动服务器
make dev
uvicorn run --host 0.0.0.0 --port 8000 mcp.server.fastmcp.cli:create_app
3. 使用示例
发送请求到 http://localhost:8000/process
:
{
"data": {
"input": "处理我的数据",
"parameters": {}
}
}
✨ 主要特性
- 提供完整的项目结构,包含项目构建、依赖管理和服务器启动脚本。
- 支持提示模板管理和对话管理接口,方便开发相关功能。
- 所有接口均支持 JSON 格式请求和响应,便于集成。
📦 安装指南
安装依赖
pip install -r requirements.txt
注意事项
⚠️ 重要提示
安装依赖前请确保已安装 setuptools
和 wheel
。项目使用 Python 3.8+
,推荐使用虚拟环境。
💻 使用示例
基础用法
发送请求到 http://localhost:8000/process
:
{
"data": {
"input": "处理我的数据",
"parameters": {}
}
}
📚 详细文档
API 文档
浏览器打开 http://localhost:8000/docs。
提示模板管理
/prompts
路由提供提示模板的增删改查功能。
对话管理
/dialogs
路由提供对话的增删改查功能。
🔧 技术细节
项目结构
your-project-name/
├── Makefile # 项目构建和运行脚本
├── setup.py # 安装依赖和可执行脚本
├── requirements.txt # 项目依赖列表
├── mcp/
│ ├── __init__.py # 包初始化文件
│ └── server/
│ ├── __init__.py # 包初始化文件
│ └── fastmcp/
│ ├── __init__.py # 包初始化文件
│ ├── prompts/ # 提示模板模块
│ │ └── sample_prompts.py # 示例提示模板实现
│ └── dialogs/ # 对话管理模块
│ └── sample_dialogs.py # 示例对话实现
支持的接口
提示模板管理接口
/prompts
:
- GET: 列出所有提示模板
- POST: 创建新提示模板
- PUT: 更新指定提示模板
- DELETE: 删除指定提示模板
对话管理接口
/dialogs
:
- POST: 创建新对话
- GET: 获取指定对话状态
- PUT: 更新指定对话内容
- DELETE: 删除指定对话
返回格式
{
"id": "string", # 唯一标识符
"status": "success" | "error",
"message": "操作成功" | "错误信息"
}
代码片段
setup.py
from setuptools import setup, find_packages
setup(
name="your_project_name",
version="0.1.0",
packages=find_packages(),
install_requires=[
"fastapi",
"uvicorn",
"python-multipart",
"numpy",
"pandas",
"langchain",
"huggingface",
"tqdm"
],
entry_points={
"console_scripts": [
"mcp-server=mcp.server.fastmcp.cli:main",
],
},
)
Makefile
PROJECT_NAME = your-project-name
VENV := .venv
all: clean
setup:
pip install -r requirements.txt
dev:
python -m venv $(VENV)
source $(VENV)/bin/activate
set PYTHONPATH=$(PYTHONPATH):$(shell pwd)
install:
pip install -e .
clean:
find . -name "*.pyc" -delete
find . -name "__pycache__" -type d -delete
mcp/server/fastmcp/cli.py
import uvicorn
from fastapi import FastAPI
from mcp.server.fastmcp.routers import main_router
def create_app():
app = FastAPI()
app.include_router(main_router.router)
return app
if __name__ == "__main__":
uvicorn.run("mcp.server.fastmcp.cli:create_app", host="0.0.0.0", port=8000, reload=True)
mcp/server/fastmcp/routers/main.py
from fastapi import APIRouter
from fastapi.responses import JSONResponse
from typing import List, Dict
router = APIRouter()
@router.get("/", response_class=JSONResponse)
async def root():
return {"message": "MCP 服务器运行中"}
@router.post("/process", response_class=JSONResponse)
async def process(data: dict) -> dict:
return {"status": "success", "result": data}
mcp/server/fastmcp/prompts/base.py
class PromptTemplate:
def __init__(self, template_str: str):
self.template = template_str
def format(self, **kwargs) -> str:
return self.template.format(**kwargs)
mcp/server/fastmcp/dialogs/sample_dialogs.py
from typing import List, Dict, Any
from langchain.schema import Dialogue
from fastapi import APIRouter
dialog_router = APIRouter()
@dialog_router.post("/dialogs", response_class=JSONResponse)
async def create_dialog(request: dict) -> Dict[str, Any]:
dialog_id = str(uuid.uuid4())
return {"id": dialog_id}
@dialog_router.get("/dialogs/{dialog_id}", response_class=JSONResponse)
async def get_dialog(dialog_id: str) -> Dict[str, Any]:
pass
@dialog_router.put("/dialogs/{dialog_id}", response_class=JSONResponse)
async def update_dialog(dialog_id: str, data: dict) -> Dict[str, Any]:
pass
@dialog_router.delete("/dialogs/{dialog_id}", response_class=JSONResponse)
async def delete_dialog(dialog_id: str) -> Dict[str, Any]:
pass
mcp/server/fastmcp/prompts/sample_prompts.py
from typing import List, Dict, Any
from langchain.prompts import PromptTemplate as LangPromptTemplate
from fastapi import APIRouter
prompt_router = APIRouter()
@prompt_router.get("/prompts", response_class=JSONResponse)
async def list_prompts() -> List[Dict[str, str]]:
pass
@prompt_router.post("/prompts", response_class=JSONResponse)
async def create_prompt(request: dict) -> Dict[str, Any]:
pass
@prompt_router.get("/prompts/{prompt_id}", response_class=JSONResponse)
async def get_prompt(prompt_id: str) -> Dict[str, Any]:
pass
@prompt_router.put("/prompts/{prompt_id}", response_class=JSONResponse)
async def update_prompt(prompt_id: str, data: dict) -> Dict[str, Any]:
pass
@prompt_router.delete("/prompts/{prompt_id}", response_class=JSONResponse)
async def delete_prompt(prompt_id: str) -> Dict[str, Any]:
pass