LangGraph代理的MCP服务器
基于FastMCP的模型上下文协议服务器,用于ChatGPT Enterprise与LangGraph代理的集成。
🎉 快速开始
最快的入门方法:
./start.sh这将:
- 检查您的LangGraph代理是否在端口2024上运行
- 在端口8000上启动MCP服务器
- 在端口3005上启动测试UI
- 打开http://localhost:3005在浏览器中
概述
此MCP服务器提供了一个标准化的接口,用于与部署在端口2024上的LangGraph代理进行交互,符合ChatGPT Enterprise集成要求。
✨ 最新动态
2025年1月-LangGraph CLI集成
- ✅ 重构为使用LangGraph CLI API体系结构
- ✅ 更新至
/runs,/runs/stream,/ok端点 - ✅ 已将消息格式更改为
{"type": "human"} - ✅ 新增3个工具:健康检查、代理状态、线程列表
- ✅ 修复了所有异步操作(不再有阻塞调用)
- ✅ 更新了web UI以匹配新的API结构
看 REFACTORING_SUMMARY.md 详细更改。
特性
- ✅ MCP协议2025-06-18 顺从的
- ✅ ChatGPT企业版 兼容(SSE传输)
- ✅ OAuth 2.0身份验证 -Google OAuth和API关键支持
- ✅ FastMCP 2.13.0+ 生产就绪部署框架
- ✅ LangGraph CLI API 整合
- ✅ 6工具:invoke_agent、stream_agent、check_system_health、check_agent_status、get_thread_state、list_threads
- ✅ 2资源:代理健康检查和服务器信息
- ✅ 提示:格式化代理查询
- ✅ Web测试UI:端口3005上的交互式测试界面
- ✅ 缺省巩固安全:生产部署的可选身份验证
安装
- 安装依赖项:
pip install -r requirements.txt- 配置环境(可选):
# Copy example configuration
cp .env.example .env
# Edit .env with your settings
# For development without auth:
OAUTH_ENABLED=false
# For production with auth:
OAUTH_ENABLED=true
GOOGLE_CLIENT_ID=your-client-id
GOOGLE_CLIENT_SECRET=your-client-secret
API_KEYS=your-api-key-1,your-api-key-2- 生成凭据(如果使用OAuth):
python generate_credentials.py看 OAUTH_SETUP.md 详细的身份验证设置。
用法
选项1:快速启动脚本(推荐)
./start.sh选项2:手动启动
启动MCP服务器:
python src/agent_mcp/mcp_server.py启动测试UI(可选):
cd web_ui && python server.py选项3:使用FastMCP CLI
python -m agent_mcp.mcp_server或者使用FastMCP CLI:
fastmcp run src/agent_mcp/mcp_server.py对于地方发展(STDIO):
fastmcp dev src/agent_mcp/mcp_server.py定制运输:
from agent_mcp.mcp_server import mcp
# HTTP transport
mcp.run(transport="http", host="0.0.0.0", port=8000, path="/mcp")
# SSE transport (for ChatGPT Enterprise)
mcp.run(transport="sse", host="0.0.0.0", port=8000)可用工具
1. invoke_agent
执行LangGraph代理的单个调用。
{
"prompt": "What is the capital of France?",
"thread_id": "optional-thread-id"
}2. stream_agent
来自LangGraph代理的流响应。
{
"prompt": "Tell me a story",
"thread_id": "optional-thread-id"
}3. get_agent_state
检索对话线程的当前状态。
{
"thread_id": "thread-id-to-query"
}认证
MCP服务器支持三种用于生产部署的身份验证方法:
1.OAuth 2.0(谷歌或Okta)
使用首选身份提供程序启用基于用户的身份验证:
谷歌OAuth:
# .env configuration
OAUTH_ENABLED=true
OAUTH_PROVIDER=google
GOOGLE_CLIENT_ID=your-client-id.apps.googleusercontent.com
GOOGLE_CLIENT_SECRET=your-client-secretOkta OAuth:
# .env configuration
OAUTH_ENABLED=true
OAUTH_PROVIDER=okta
OKTA_DOMAIN=your-domain.okta.com
OKTA_CLIENT_ID=your-okta-client-id
OKTA_CLIENT_SECRET=your-okta-client-secretOAuth端点:
GET /auth/login-启动OAuth流GET /auth/callback-OAuth回调处理程序GET /auth/logout-注销GET /auth/status-检查身份验证状态
快速入门指南:
2.API密钥认证
使用API密钥进行服务到服务身份验证:
# Generate API keys
python generate_credentials.py
# Add to .env
API_KEYS=key1,key2,key3使用API密钥:
# cURL
curl -H "X-API-Key: your-api-key" http://localhost:8000/sse
# Python
headers = {"X-API-Key": "your-api-key"}测试身份验证
# Test OAuth setup
python test_oauth.py
# Or test manually
curl http://localhost:8000/health # Public endpoint
curl -H "X-API-Key: your-key" http://localhost:8000/sse # Protected有关详细的设置说明,请参阅 OAUTH_SETUP.md
资源
agent://health-代理人健康检查agent://info-代理功能和元数据
提示
agent_query_prompt-为代理设置查询格式
ChatGPT企业集成
此服务器专为ChatGPT Enterprise集成而设计,具有:
- 苏格兰和南方能源公司运输:实时流媒体的默认传输
- MCP协议2025-06-18:最新稳定协议版本
- 正确的工具模式:根据Python类型提示自动生成
- 上下文支持:日志记录和进度报告
- 错误处理:全面的错误响应
📚 企业应用程序和连接器文档
在与ChatGPT Enterprise“应用程序和连接器”集成时遇到问题?
我们创建了全面的文档来帮助您:
- 从这里开始:执行摘要 -问题和解决方案的快速概述
- 完整研究报告 -详细的技术分析
- 故障排除检查表 -逐步诊断指南
- 代码修复指南 -准备实施改进
- 企业管理指南 -适用于工作区管理员
- 视觉架构指南 -图表和流程图
- 完整文档索引 -所有资源集中在一个地方
太长,读不下去了:如果您的OpenAPI服务器在自定义GPT中工作,但在企业“应用程序和连接器”中不工作,则可能是由于企业工作区域限制。联系您的企业管理员进入白名单 mcp.marananapp.com 在GPT操作设置中。
ChatGPT配置
添加到您的ChatGPT Enterprise MCP配置中:
{
"mcpServers": {
"langgraph-agent": {
"url": "http://your-server:8000/sse",
"transport": "sse"
}
}
}测试
Web UI测试工具
我们提供了一个漂亮的基于web的UI来测试您的MCP服务器和LangGraph代理:
# Start the test UI server
cd web_ui
python server.py然后打开http://localhost:3005在浏览器中:
- 测试MCP服务器连接
- 测试LangGraph代理连接
- 使用自定义提示调用代理
- 实时流式响应
- 查看活动日志
看 web_ui/README.md 了解详情。
单元测试
运行测试:
pytest tests/test_mcp_server.py -v运行所有测试:
pytest发展
项目结构
agent-mcp-py/
├── src/
│ └── agent_mcp/
│ ├── __init__.py
│ └── mcp_server.py # FastMCP server implementation
├── tests/
│ ├── test_mcp_server.py # MCP server tests
│ └── test_*.py # Other tests
├── requirements.txt # Production dependencies
├── requirements-dev.txt # Development dependencies
└── README.md添加新工具
from fastmcp import Context
@mcp.tool()
async def my_tool(param: str, ctx: Context = None) -> dict:
"""Tool description for ChatGPT."""
if ctx:
await ctx.info(f"Processing: {param}")
# Your logic here
return {"result": "success"}添加资源
@mcp.resource("custom://resource")
async def my_resource() -> str:
"""Resource description."""
return "Resource content"建筑
┌─────────────────┐
│ ChatGPT │
│ Enterprise │
└────────┬────────┘
│ MCP/SSE
│
┌────────▼────────┐
│ FastMCP │
│ Server │
│ (Port 8000) │
└────────┬────────┘
│ HTTP
│
┌────────▼────────┐
│ LangGraph │
│ Agent │
│ (Port 2024) │
└─────────────────┘生产部署
通过身份验证
from fastmcp.server.auth.providers.google import GoogleProvider
auth = GoogleProvider(
client_id="your-client-id",
client_secret="your-client-secret",
base_url="https://your-domain.com"
)
mcp = FastMCP(
"LangGraph Agent Server",
auth=auth
)Docker部署
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY src/ ./src/
CMD ["python", "-m", "agent_mcp.mcp_server"]许可证
麻省理工学院
