MCP Forge
什么是MCP服务器?
MCP服务器是一个标准化的工具开发框架,它通过集成FastAPI和FastAPI-MCP,将传统的API接口转换为可被AI调用的MCP工具。如何使用MCP服务器?
只需定义服务接口、实现业务逻辑、创建API端点并启用MCP工具即可快速部署AI工具。适用场景
适用于AI工具开发团队、希望将现有API转换为AI工具的开发者以及实施微服务架构的企业。主要功能
自动转换传统API为MCP工具通过FastAPI-MCP自动将API端点转换为可被AI调用的MCP工具。
接口与实现分离支持清晰分离接口定义和具体实现,便于测试和环境切换。
依赖注入设计利用FastAPI的依赖注入机制实现灵活的组件组合和解耦。
完整的开发流水线提供从开发到测试再到部署的全面支持。
示例驱动的文档通过实践示例演示最佳实践,加速上手过程。
优势与局限性
优势
快速构建高质量AI工具
标准化MCP工具开发流程
易于维护和扩展
局限性
需要一定的Python基础
对复杂业务逻辑的支持可能有限
如何使用
安装依赖
使用uv作为包管理器安装依赖并设置开发环境。
启动开发服务器
启动开发服务器以运行示例服务。
访问API文档
在浏览器中访问http://localhost:5000/docs查看API文档。
连接MCP端点
使用MCP客户端连接到http://localhost:5000/mcp。
使用案例
搜索项目定义服务接口、实现Mock或Real服务、创建API端点并通过FastAPI-MCP生成MCP工具。
情感分析定义预测情感的API端点并通过FastAPI-MCP生成MCP工具。
常见问题
如何安装MCP服务器?
使用uv作为包管理器安装依赖并设置开发环境。
如何启动开发服务器?
运行命令'make dev'启动开发服务器。
如何验证API是否正常工作?
在浏览器中访问http://localhost:5000/docs查看API文档。
相关资源
FastAPI官方文档
FastAPI框架的详细文档。
FastAPI-MCP GitHub仓库
FastAPI-MCP工具的开源代码库。
安装
复制以下命令到你的Client进行配置
注意:您的密钥属于敏感信息,请勿与任何人分享。
🚀 使用 FastAPI 和 FastAPI-MCP 开发 MCP 服务器框架
本项目借助 FastAPI 和 FastAPI-MCP 构建 MCP 服务器框架,可在多个云平台协调管理计算任务,提供丰富 API 接口,支持多协议与集成方案,适用于复杂云环境。
🚀 快速开始
要开始使用 FastAPI 和 FastAPI-MCP 创建一个基本的 MCP 服务器,请按照以下步骤操作:
-
安装必要的依赖项:
pip install fastapi fastapi_mcp uvicorn
-
创建主应用文件 (main.py):
from fastapi import FastAPI from fastapi_mcp import FastApiMCP app = FastAPI() # 定义一个标准的 FastAPI 端点 @app.post("/predict", operation_id="predict_sentiment") async def predict_sentiment(text: str): return {"sentiment": "positive", "confidence": 0.92} # 创建并挂载 MCP 服务,自动将上面的端点转换为 MCP 工具 mcp = FastApiMCP( app, name="sentiment-analysis", description="情感分析服务", base_url="http://localhost:5000", include_operations=["predict_sentiment"] ) # 挂载 MCP 服务到指定路径 mcp.mount(mount_path="/mcp") if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)
-
运行应用:
python main.py
-
访问服务:
- 访问 FastAPI 端点:
http://localhost:5000/predict
- 访问 MCP 工具:
http://localhost:5000/mcp
- 访问 FastAPI 端点:
✨ 主要特性
核心功能
- 跨云平台兼容性:支持 AWS、GCP、Azure 等主流云服务提供商。
- 负载均衡与故障转移:自动分配任务到不同的云实例,确保高可用性。
- 日志与监控集成:内置对 Prometheus 和 Grafana 的支持,方便实时监控和分析。
- 安全认证:支持 OAuth2、JWT 等多种身份验证机制,保障数据安全。
高级功能
- 分布式事务管理:确保跨云服务的事务一致性。
- 缓存与加速:集成 Redis 或 Memcached 进行结果缓存,提升响应速度。
- 动态扩展:根据负载自动调整资源使用,优化成本。
- 智能路由:基于地理位置或策略分配请求到最佳可用节点。
📦 安装指南
环境要求
- 操作系统:Linux (推荐) 或 macOS
- Python 版本:3.6+
- 依赖管理工具:poetry/pipenv (可选)
- 额外工具:Docker, kubectl (如需容器化部署)
安装步骤
-
克隆项目仓库:
git clone https://github.com/yourusername/mcp-server.git cd mcp-server
-
创建虚拟环境并安装依赖:
python -m venv .venv source .venv/bin/activate # 在 macOS/Linux 下 pip install -r requirements.txt
-
配置项目:
- 修改
config.py
中的云服务提供商凭证和 API 端点。 - 更新日志记录配置,确保输出到合适的位置。
- 修改
-
启动服务器:
uvicorn main:app --host 0.0.0.0 --port 8000
另一种安装方式
安装依赖
pip install mcp-server
配置文件
创建 config.yaml
文件,如下所示:
cloud_providers:
- name: AWS
region: us-east-2
access_key: YOUR_AWS_KEY
secret_key: YOUR_AWS_SECRET
- name: GCP
region: europe-west1
project_id: your-gcp-project
key_path: path/to/service-account.json
- name: Azure
region: eastus
subscription_id: your_azure_subscription_id
client_id: your_azure_client_id
client_secret: your_azure_client_secret
启动服务器
python3 -m mcp_server.run --config config.yaml
💻 使用示例
基础用法
情感分析服务
# app/endpoints/sentiment_endpoints.py
from fastapi import APIRouter, Depends
from app.services.sentiment_service import SentimentService
router = APIRouter()
@router.post("/sentiment")
async def analyze_sentiment(text: str, service: SentimentService = Depends(SentimentService)):
sentiment = await service.analyze(text)
return {"sentiment": sentiment}
停车场搜索服务
# app/endpoints/parking_endpoints.py
from fastapi import APIRouter, Depends
from app.services.parking_service import ParkingService
router = APIRouter()
@router.get("/parking/nearby")
async def find_nearby_parkings(latitude: float, longitude: float, radius: int,
service: ParkingService = Depends(ParkingService)):
result = await service.find_nearby_parkings(latitude, longitude, radius)
return {"data": result}
高级用法
个性化推荐系统
# app/endpoints/recommendation_endpoints.py
from fastapi import APIRouter, Depends
from app.services.recommendation_service import RecommendationService
router = APIRouter()
@router.post("/recommendations")
async def get_recommendations(user_id: int, service: RecommendationService = Depends(RecommendationService)):
recommendations = await service.get Recommendations(user_id)
return {"recommendations": recommendations}
分布式事务管理
from mcp_server.transaction import TransactionManager
tm = TransactionManager()
with tm.context():
# 执行跨云操作
pass
智能路由与负载均衡
class SmartRouter:
def route_request(self, request):
# 返回目标云服务提供商和区域
pass
动态扩展
# Kubernetes Horizontal Pod Autoscaler 配置
apiVersion: autoscaling/v1
kind: HorizontalPodAutoscaler
metadata:
name: mcp-server-hpa
spec:
scaleRef:
kind: Deployment
name: mcp-server-deployment
apiVersion: apps/v1
minReplicas: 2
maxReplicas: 5
targetCPUUtilizationPercentage: 80
📚 详细文档
项目结构
一个典型的 MCP 服务器项目可能包含以下文件和目录:
mcp_server/
├── main.py # 主应用文件
├── app/ # FastAPI 应用模块
│ ├── models.py # 数据模型定义
│ ├── services/ # 服务类(如数据库连接、业务逻辑)
│ └── endpoints/ # API 端点实现
├── config/ # 配置管理
│ ├── settings.py # 应用设置
│ └── environment.py # 环境变量管理
└── tests/ # 测试用例
├── test_endpoints.py # 单元测试
└── integration_tests.py # 集成测试
基础功能实现
创建服务类
# app/services/parking_service.py
class ParkingService:
def __init__(self, db):
self.db = db
async def find_nearby_parkings(self, latitude: float, longitude: float, radius: int) -> list:
# 查询数据库以获取附近的停车场
pass
实现 API 端点
# app/endpoints/parking_endpoints.py
from fastapi import APIRouter, Depends
from app.services.parking_service import ParkingService
from config.environment import get_db
router = APIRouter()
@router.get("/parking/nearby")
async def find_nearby_parkings(latitude: float, longitude: float, radius: int,
service: ParkingService = Depends(ParkingService)):
db = next(get_db())
result = await service.find_nearby_parkings(latitude, longitude, radius)
return {"data": result}
配置管理
# config/settings.py
class Settings:
def __init__(self):
self.debug = True # 调试模式,默认为真
@property
def database_url(self) -> str:
return "sqlite:///./test.db"
@property
def secret_key(self) -> str:
return "your-secret-key-here"
高级功能实现
依赖注入
FastAPI 的依赖注入机制非常强大,可以用于管理各种依赖项。
# config/database.py
def get_db():
db = connect_to_db()
try:
yield db
finally:
db.close()
@app.get("/items/")
async def get_items(db=Depends(get_db)):
return db.query(Item).all()
环境适应
# app/services/database_service.py
class DatabaseService:
def __init__(self, settings):
self.settings = settings
self.db = connect_to_db(settings.database_url)
async def query(self, sql: str) -> list:
pass
日志记录和监控
集成日志库(如 logging)和监控工具(如 Prometheus、Grafana)可以提供更详细的运行时信息。
常用功能示例
情感分析服务
# app/endpoints/sentiment_endpoints.py
from fastapi import APIRouter, Depends
from app.services.sentiment_service import SentimentService
router = APIRouter()
@router.post("/sentiment")
async def analyze_sentiment(text: str, service: SentimentService = Depends(SentimentService)):
sentiment = await service.analyze(text)
return {"sentiment": sentiment}
停车场搜索服务
# app/endpoints/parking_endpoints.py
from fastapi import APIRouter, Depends
from app.services.parking_service import ParkingService
router = APIRouter()
@router.get("/parking/nearby")
async def find_nearby_parkings(latitude: float, longitude: float, radius: int,
service: ParkingService = Depends(ParkingService)):
result = await service.find_nearby_parkings(latitude, longitude, radius)
return {"data": result}
个性化推荐系统
# app/endpoints/recommendation_endpoints.py
from fastapi import APIRouter, Depends
from app.services.recommendation_service import RecommendationService
router = APIRouter()
@router.post("/recommendations")
async def get_recommendations(user_id: int, service: RecommendationService = Depends(RecommendationService)):
recommendations = await service.get Recommendations(user_id)
return {"recommendations": recommendations}
常见问题与故障排除
- 问:如何处理 API 端点的错误? 答:可以使用 FastAPI 的异常处理机制,定义自定义异常并将其注册到应用中。
- 问:如何实现速率限制和认证?
答:可以集成第三方库如
fastapi-limiter
和python-jose
来实现这些功能。
性能优化与扩展
使用缓存技术
# config/caching.py
from fastapi import Request
from typing import Any
from fastapi.responses import Response
class CacheInterceptor:
def __init__(self, cache):
self.cache = cache
async def intercept_request(self, request: Request) -> Any:
key = f"{request.method}_{request.url}"
value = await self.cache.get(key)
if value is not None:
return Response(value)
async def intercept_response(self, response: Response) -> None:
key = f"{response.request.method}_{response.request.url}"
value = str(response.body(), encoding="utf-8")
await self.cache.set(key, value)
分布式事务管理
# config/transaction.py
from fastapi import Depends
from typing import Any
import asyncio
class TransactionManager:
def __init__(self):
pass
async def begin(self):
pass
async def commit(self):
await asyncio.sleep(1) # 示例事务提交时间
async def rollback(self):
pass
实时监控与日志记录
# config/monitoring.py
import logging
from fastapi import FastAPI
from prometheus_fastapi_instrumentator import Instrumentator
app = FastAPI()
Instrumentator.instrument(app)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
项目部署与维护
使用 Docker 部署
创建 Dockerfile
:
FROM python:3.8-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "--host", "0.0.0.0", "--port", "8000", "main:app"]
构建镜像并运行:
docker build -t mcp-server .
docker run -p 8000:8000 mcp-server
使用 Kubernetes 扩展
创建 Deployment
和 Service
文件:
# deployment.yml
apiVersion: apps/v1
kind: Deployment
metadata:
name: mcp-server-deployment
spec:
replicas: 3
selector:
matchLabels:
app: mcp-server
template:
metadata:
labels:
app: mcp-server
spec:
containers:
- name: mcp-server
image: mcp-server:latest
ports:
- containerPort: 8000
# service.yml
apiVersion: v1
kind: Service
metadata:
name: mcp-server-service
spec:
type: LoadBalancer
ports:
- port: 80
targetPort: 8000
selector:
app: mcp-server
应用配置:
kubectl apply -f deployment.yml -f service.yml
使用 Prometheus 监控
配置 Prometheus
监控 Kubernetes 集群中的 MCP Server:
# prometheus.yml
scrape_configs:
- job_name: 'mcp-server'
static_configs:
- targets: ['<kubernetes-master-ip>:80']
使用 Grafana 可视化
创建 Dashboard
来展示监控数据,集成不同的图表和警报规则。
🔧 技术细节
核心功能技术细节
- 跨云平台兼容性:通过封装不同云服务提供商的 API 接口,实现对 AWS、GCP、Azure 等主流云服务的支持。
- 负载均衡与故障转移:采用算法根据云实例的负载和可用性,自动分配任务,同时在某个实例出现故障时,能快速将任务转移到其他可用实例。
- 日志与监控集成:内置对 Prometheus 和 Grafana 的支持,通过在代码中集成相应的 SDK,实现实时数据的采集和展示。
- 安全认证:支持 OAuth2、JWT 等多种身份验证机制,在 API 接口层进行身份验证和授权,保障数据安全。
高级功能技术细节
- 分布式事务管理:集成
atom
或orchestrator
等工具,确保跨云服务的事务一致性。 - 缓存与加速:集成 Redis 或 Memcached 进行结果缓存,通过在代码中添加缓存逻辑,提升响应速度。
- 动态扩展:根据负载自动调整资源使用,通过集成 Kubernetes 或 Elastic Beanstalk,实现计算资源的自动扩展。
- 智能路由:基于地理位置或策略分配请求到最佳可用节点,通过自定义路由逻辑,根据实时指标和预设策略进行请求分发。
📄 许可证
文档中未提及相关许可证信息。
最佳实践
- 模块化设计:将功能拆分成独立的模块或服务,便于维护和扩展。
- 依赖管理:使用工具如
poetry
或pipenv
管理 Python 依赖,确保环境一致性。 - 配置管理:使用配置文件而不是硬编码值,方便在不同环境中部署。
- 日志记录:集成结构化日志记录库如
logging
或sentry
,便于调试和分析问题。 - 测试驱动开发:编写单元测试、集成测试和端到端测试,确保代码质量。
项目总结
通过以上步骤,您可以构建一个高效可靠的 MCP Server。从模块化的 API 设计到性能优化和扩展,每个细节都至关重要。使用 Docker 和 Kubernetes 进行容器化部署,并结合 Prometheus 和 Grafana 进行监控和可视化,将使您的服务更加稳定和易于管理。

Notte Browser
已认证
Notte是一个开源的全栈网络AI代理框架,提供浏览器会话、自动化LLM驱动的代理、网页观察与操作、凭证管理等功能,旨在将互联网转化为代理友好的环境,并通过自然语言描述网站结构,降低LLM的认知负担。
796
4.5分

Search1api
Search1API MCP Server是一个基于Model Context Protocol (MCP)的服务器,提供搜索和爬取功能,支持多种搜索服务和工具。
TypeScript
463
4分

Duckduckgo MCP Server
已认证
DuckDuckGo搜索MCP服务器,为Claude等LLM提供网页搜索和内容抓取服务
Python
1.2K
4.3分

Bing Search MCP
一个用于集成微软Bing搜索API的MCP服务器,支持网页、新闻和图片搜索功能,为AI助手提供网络搜索能力。
Python
358
4分

MCP Alchemy
已认证
MCP Alchemy是一个连接Claude Desktop与多种数据库的工具,支持SQL查询、数据库结构分析和数据报告生成。
Python
433
4.2分

Postgresql MCP
一个基于FastMCP库的PostgreSQL数据库MCP服务,提供对指定表的CRUD操作、模式检查和自定义SQL查询功能。
Python
160
4分

Agentic Radar
Agentic Radar是一个用于分析和评估代理系统的安全扫描工具,帮助开发者、研究人员和安全专家理解代理系统的工作流程并识别潜在漏洞。
Python
594
5分

MCP Scan
MCP-Scan是一款针对MCP服务器的安全扫描工具,用于检测提示注入、工具污染和跨域升级等常见安全漏洞。
Python
650
5分
精选MCP服务推荐

Duckduckgo MCP Server
已认证
DuckDuckGo搜索MCP服务器,为Claude等LLM提供网页搜索和内容抓取服务
Python
1.2K
4.3分

Figma Context MCP
Framelink Figma MCP Server是一个为AI编程工具(如Cursor)提供Figma设计数据访问的服务器,通过简化Figma API响应,帮助AI更准确地实现设计到代码的一键转换。
TypeScript
6.9K
4.5分

Firecrawl MCP Server
Firecrawl MCP Server是一个集成Firecrawl网页抓取能力的模型上下文协议服务器,提供丰富的网页抓取、搜索和内容提取功能。
TypeScript
4.3K
5分

Exa Web Search
已认证
Exa MCP Server是一个为AI助手(如Claude)提供网络搜索功能的服务器,通过Exa AI搜索API实现实时、安全的网络信息获取。
TypeScript
2.0K
5分

Minimax MCP Server
MiniMax Model Context Protocol (MCP) 是一个官方服务器,支持与强大的文本转语音、视频/图像生成API交互,适用于多种客户端工具如Claude Desktop、Cursor等。
Python
1.0K
4.8分

Context7
Context7 MCP是一个为AI编程助手提供实时、版本特定文档和代码示例的服务,通过Model Context Protocol直接集成到提示中,解决LLM使用过时信息的问题。
TypeScript
5.6K
4.7分

Baidu Map
已认证
百度地图MCP Server是国内首个兼容MCP协议的地图服务,提供地理编码、路线规划等10个标准化API接口,支持Python和Typescript快速接入,赋能智能体实现地图相关功能。
Python
958
4.5分

Edgeone Pages MCP Server
EdgeOne Pages MCP是一个通过MCP协议快速部署HTML内容到EdgeOne Pages并获取公开URL的服务
TypeScript
398
4.8分
智启未来,您的人工智能解决方案智库
简体中文