监控和可观察性MCP服务器
使用Langfuse对系统进行全面监控和可观察性的模型上下文协议(MCP)服务器。
🎯 这有什么作用
此MCP服务器允许您:
- 监视器 实时监控您的所有代理
- 轨道 性能指标(延迟、成本、令牌使用)
- 调试 执行失败,有详细的痕迹
- 分析 跨时间段的代理性能
- 比较 通过元数据过滤器实现不同的代理版本
- 管理 成本和设置预算警报
- 可视化 代理工作流
快速开始
1.先决条件
- Python 3.11或更高版本
- Langfuse账户(在这里注册)
- 装有Langfuse的试剂
2.安装
# Install via pip
pip install -r requirements.txt
# Or install from source
git clone https://github.com/yourusername/langfuse-mcp-python.git
cd langfuse-mcp-python
pip install -e .3.配置
创建一个 .env 使用您的Langfuse凭据文件:
cp .env.example .env
# Edit .env and add your credentials你的 .env 应该看起来像:
LANGFUSE_PUBLIC_KEY=pk-lf-xxxxx
LANGFUSE_SECRET_KEY=sk-lf-xxxxx
LANGFUSE_HOST=https://cloud.langfuse.com4.以流式HTTP(URL)运行
如果你想要一个适用于所有工具的Streamable HTTP URL,请使用Streamable HTTP传输运行服务器:
python -m langfuse_mcp_python --transport streamable-http --host 127.0.0.1 --port 8000 --path /mcppython -m langfuse_mcp_python --transport sse --host 127.0.0.1 --port 8000然后,您可以将任何兼容Streamable HTTP的MCP客户端连接到:
http://127.0.0.1:8000/mcp如果您使用的是Claude Desktop或Cursor,请保留默认值 stdio 运输在他们的配置。
4b。设置MCP客户端
适用于克劳德桌面
添加 claude_desktop_config.json:
{
"mcpServers": {
"langfuse-monitor": {
"command": "uvx",
"args": ["--python", "3.11", "langfuse-mcp-python"],
"env": {
"LANGFUSE_PUBLIC_KEY": "pk-lf-xxxxx",
"LANGFUSE_SECRET_KEY": "sk-lf-xxxxx",
"LANGFUSE_HOST": "https://cloud.langfuse.com"
}
}
}
}对于光标
添加 .cursor/mcp.json:
{
"mcpServers": {
"langfuse-monitor": {
"command": "python",
"args": ["-m", "langfuse_mcp_python"],
"env": {
"LANGFUSE_PUBLIC_KEY": "pk-lf-xxxxx",
"LANGFUSE_SECRET_KEY": "sk-lf-xxxxx"
}
}
}
}5.为你的代理人提供工具
确保你的代理人向Langfuse发送追踪信息:
from langfuse.langchain import CallbackHandler
from langgraph.graph import StateGraph
# Create Langfuse callback handler
langfuse_handler = CallbackHandler(
public_key="pk-lf-xxxxx",
secret_key="sk-lf-xxxxx",
host="https://cloud.langfuse.com"
)
# Create your agent
workflow = StateGraph(AgentState)
workflow.add_node("planner", planner_node)
workflow.add_node("executor", executor_node)
app = workflow.compile()
# Run with Langfuse monitoring
result = app.invoke(
{"input": "user query"},
config={
"callbacks": [langfuse_handler],
"metadata": {
"agent_name": "my_planner_agent",
"version": "v1.0"
}
}
)项目结构
src/langfuse_mcp_python/server.pyCLI入口点和stdio传输src/langfuse_mcp_python/http_server.py流式HTTP和SSE传输src/langfuse_mcp_python/utils/tool_registry.py工具设置和注册src/langfuse_mcp_python/tools/工具实施和规格src/langfuse_mcp_python/integrations/langfuse_client.pyAPI客户src/langfuse_mcp_python/core/base_tool.py共享缓存和指标
可用工具
监控和分析
watch_agents监控活动代理get_trace按ID获取跟踪analyze_performance随时间变化的综合性能get_metrics聚合指标(延迟、成本、令牌)
分数和评价
get_scores获取分数submit_score创建分数get_score_configs列出分数配置
提示
get_prompts列表提示create_prompt创建提示delete_prompt删除提示
会话
get_sessions列出会话
数据集
get_datasets列出数据集create_dataset创建数据集create_dataset_item将项目添加到数据集
模型
get_models列出型号create_model创建一个模型delete_model删除模型
评论
get_comments列出评论add_comment添加评论
痕迹
delete_trace删除跟踪
注释队列
get_annotation_queues列出注释队列create_annotation_queue创建队列get_queue_items列出队列项目resolve_queue_item解析队列项
Blob存储集成
get_blob_storage_integrations列出集成upsert_blob_storage_integration创建或更新集成get_blob_storage_integration_status获取集成状态delete_blob_storage_integration删除集成
LLM连接
get_llm_connections列出连接upsert_llm_connection创建或更新连接
项目
get_projects列出项目create_project创建项目update_project更新项目delete_project删除项目
例子: watch_agents
实时监控所有活动代理。
例子:
Show me all active agents from the last hour答复:
Active Agent Monitoring (last_1h)
Total Traces Found: 15
Showing: Top 10 traces
1. research_agent (Trace: trace-abc12...)
- Status: completed
- Session: session-xyz
- Started: 2026-03-19T10:25:00Z
- Latency: 1250ms
- Tokens: 3420
- Cost: $0.0234高级用法
过滤剂
Watch only my research_agent and planner_agent from the last 24 hours性能分析
Analyze performance of my planner_agent over the last 24 hours成本监控
Show cost breakdown by agent for the last week深度调试
Show trace details for trace-abc123建筑
MCP Client (Claude, Cursor, etc.)
-> Langfuse MCP Server (stdio/HTTP)
-> Langfuse API
-> Langfuse Platform
-> Your Langfuse Agents安全最佳实践
- 从不提交凭据 -使用环境变量
- 旋转API键 定期
- 使用只读密钥 在可能的情况下
- 启用速率限制 生产中
- 屏蔽敏感数据 在痕迹中
监控工作流程示例
每日代理人健康检查
- 检查活性剂:
watch_agents - 审查绩效:
analyze_performance - 检查成本:
get_metrics - 调查故障:
get_trace
代理优化周期
- 建立基线:
analyze_performance当前版本元数据 - 部署具有不同元数据的新版本
- 通过运行来比较版本
analyze_performance使用版本筛选器 - 做出数据驱动的部署决策
成本控制
- 跟踪成本:
get_metrics按代理分组 - 识别昂贵的代理商
- 优化高成本运营
- 随着时间的推移追踪节省情况
故障排除
MCP服务器未连接
- 检查环境变量是否设置正确
- 验证Langfuse API密钥是否有效
- 确保已安装Python 3.11+
- 检查日志:
tail -f ~/.mcp/logs/langfuse-monitor.log
未发现任何痕迹
- 验证代理是否使用Langfuse进行检测
- 检查
langfuse_handler传递给代理调用 - 确保元数据包括
agent_name - 验证时间窗口是否合适
高延迟
- 减少提取的痕迹数量(使用过滤器)
- 启用缓存:
CACHE_ENABLED=true - 使用“最小”深度获取痕迹细节
- 考虑对大型数据集进行批处理
贡献
欢迎投稿!拜托:
- 分叉存储库
- 创建要素分支
- 添加新功能的测试
- 提交拉取请求
许可证
MIT许可证-有关详细信息,请参阅许可证文件
致谢
路线图
- \[x\] 核心监控工具
- \[x\] 性能分析
- \[x\] 成本跟踪
- \[x\] 调试实用程序
- \[\]实时流媒体更新
- \[\]自定义警报系统
- \[\]预测分析
- \[\]A/B测试支持
- \[\]多项目支持
- \[\]导出到数据仓库
______________________________________________________________________
版本: 1.0.0\ 最后更新2026年3月23日\ 状态:生产就绪
