MCP Observer SDK

MCP观察者SDK

](https://badge.fury.io/py/mcp-observer) ](https://pypi.org/project/mcp-observer/) 

一个轻量级的、基于装饰器的可观察性SDK 模型上下文协议（MCP） 工具。只需一行代码，即可为您的MCP服务器添加全面的遥测和见解。

特性

• 零摩擦集成：使用简单的装饰器添加可观察性
• 开放遥测支持：内置跟踪、度量和分布式上下文传播
• 隐私第一：具有双重同意系统的可配置I/O跟踪
• 通用兼容性：适用于所有MCP工具功能签名
• 双存储器：持久存储，用于分析+通过OpenTetry实时流式传输
• 会话跟踪：自动会话和请求关联
• 跑步追踪：使用基于超时的生命周期对工具调用进行自动会话级分组

安装

来自PyPI

pip install mcp-observer

使用紫外线（推荐）

uv pip install mcp-observer

为了发展

# Clone the repository
git clone https://github.com/yourusername/mcp-observer-sdk.git
cd mcp-observer-sdk

# Install in development mode
pip install -e ".[dev]"

快速开始

from mcp_observer import MCPObserver
from fastmcp import FastMCP, Context

# Initialize your MCP server
mcp = FastMCP("MyServer")

# Initialize the observer (project is automatically determined from your API key)
observer = MCPObserver(
    name="MyServer",
    version="1.0.0",
    api_key="your-generated-api-key"
)

# Decorate your tools - IMPORTANT: Include Context parameter for run tracking
@mcp.tool()
@observer.track(track_io=True)
async def my_tool(data: dict, ctx: Context = None) -> dict:
    # Your tool logic here
    # The ctx parameter enables automatic session and run tracking
    return {"result": "success"}

💡 专业提示：始终包括 ctx: Context = None 工具中的参数，以启用正确的会话和运行跟踪。没有它，每个工具调用都将作为单独的运行进行跟踪。

例子

示例1：简单工具（无上下文）

from mcp_observer import MCPObserver
from fastmcp import FastMCP

mcp = FastMCP("MathServer")
observer = MCPObserver(
    name="MathServer",
    version="1.0.0",
    api_key="your-api-key"
)

@mcp.tool(name="adder", description="Add two numbers")
@observer.track(track_io=True)
async def add(a: int, b: int) -> int:
    """Add two numbers together"""
    return a + b

示例2：带有会话上下文的工具

from mcp_observer import MCPObserver
from fastmcp import FastMCP, Context
from typing import Optional

mcp = FastMCP("EchoServer")
observer = MCPObserver(
    name="EchoServer",
    version="1.0.0",
    api_key="your-api-key"
)

@mcp.tool(name="echo", description="Echo the input with session tracking")
@observer.track(track_io=True)
async def echo(message: str, ctx: Optional[Context] = None) -> str:
    """Echo input string with session ID"""
    if ctx:
        return f"{message} (Session: {ctx.session_id})"
    return message

示例3：仅指纹跟踪（默认）

对于处理敏感数据的工具，省略 track_io=True 仅存储指纹：

@mcp.tool(name="process_sensitive_data")
@observer.track()  # Only fingerprints stored, no full I/O
async def process_sensitive_data(user_data: dict) -> dict:
    """Process sensitive user data"""
    # Only metadata is logged, not the actual data
    return {"status": "processed"}

跑步追踪

跑动 提供会话内工具调用的自动会话级分组。运行代表一个逻辑任务或对话，并在一段时间不活动后自动关闭。

运作原理

当你包括一个 Context 工具中的参数：

1. 第一次工具调用 在会议中→ 创建新跑步记录
2. 后续通话 30秒内→ 重复使用相同的运行（分组在一起）
3. 30秒不活动后 → 上一次运行结束，下一次调用创建新的运行
4. 无上下文 → 每个调用都是自己的运行（无分组）

配置

# Default: Run tracking enabled with 30s timeout
observer = MCPObserver(
    name="MyServer",
    version="1.0.0",
    api_key="your-api-key"
)

# Custom timeout (60 seconds)
observer = MCPObserver(
    name="MyServer",
    version="1.0.0",
    api_key="your-api-key",
    run_timeout_seconds=60.0
)

# Disable run tracking (not recommended for agent use cases)
observer = MCPObserver(
    name="MyServer",
    version="1.0.0",
    api_key="your-api-key",
    run_aware=False
)

为什么要使用跑步追踪？

运行跟踪对于理解代理行为至关重要：

• 会话分析：查看哪些工具调用属于同一对话
• 性能监控：跟踪多步骤任务的端到端延迟
• 调试：跟踪相关工具调用之间的错误传播
• 使用情况分析：了解代理如何组合工具

最佳实践

✅ 做：始终包括 ctx: Context = None 在您的工具签名中

@mcp.tool()
@observer.track(track_io=True)
async def my_tool(query: str, ctx: Context = None) -> str:
    # Proper run tracking
    return "result"

❌ 不要：省略面向代理工具的Context参数

@mcp.tool()
@observer.track(track_io=True)
async def my_tool(query: str) -> str:
    # No run grouping - each call is isolated
    return "result"

MCP观察器上的参数：

构造函数

• name：服务器或应用程序的名称（这是日志记录中显示的名称）
• version：服务器/应用程序的版本
• api_key：用于身份验证的API密钥（项目由该密钥自动确定）
• run_aware：启用运行跟踪（默认值： True)
• run_timeout_seconds：关闭运行的非活动超时时间（秒）（默认值： 30.0)
• otlp_endpoint：用于分布式跟踪的可选OTLP收集器端点
• enable_console_export：启用控制台调试输出（默认值： False)
• logger：自定义记录器实例（可选）

装饰师： @observer.track()

• track_io （bool）：如果为True，则启用完整的输入/输出跟踪（需要项目同意）

- 默认值：False（仅存储指纹）

高级功能

开放遥测集成

SDK自动与OpenTetry集成，用于分布式跟踪和指标：

observer = MCPObserver(
    name="MyServer",
    version="1.0.0",
    api_key="your-api-key",
    otlp_endpoint="http://localhost:4317",  # Optional: Send to OTLP collector
    enable_console_export=True  # Optional: Enable console debugging
)

环境变量：

• OTEL_EXPORTER_OTLP_ENDPOINT：配置OTLP端点以进行生产跟踪
• OTEL_CONSOLE_EXPORT：设置为 "true" 启用控制台跨度/度量导出

自动度量：

• mcp.tool.calls：工具调用计数器
• mcp.tool.duration：工具执行时间柱状图（ms）
• mcp.tool.errors：刀具错误计数器

自动跨度：

• 每个工具调用都会创建一个名为 mcp.tool.{function_name}
• 跨度包括：call_id、session_id、延迟、状态、错误详细信息

跟踪策略系统

SDK使用双重同意系统进行完整的I/O跟踪：

1. 开发商声明安全：使用 @observer.track(track_io=True)
2. 项目管理员启用：后端API控制每到一个策略
3. 结果已缓存：缓存策略响应1小时（可配置）

这确保了只有在开发人员和管理员都同意的情况下才能存储敏感数据。

自定义日志记录

通过您自己的日志记录器与现有的日志基础设施集成：

import logging

my_logger = logging.getLogger("MyApp")
my_logger.setLevel(logging.DEBUG)

observer = MCPObserver(
    name="MyServer",
    version="1.0.0",
    api_key="your-api-key",
    logger=my_logger
)

运行示例

# Install dependencies
uv pip install -e ".[dev]"

# Run the example server
python tests/simple_example.py

API 参考

MCPObserver(name, version, api_key, **kwargs)

初始化MCP服务器的观察器。

参数：

• name （str）：您的服务器/应用程序名称
• version （str）：您的服务器/应用程序版本
• api_key （str）：身份验证密钥（项目是自动确定的）
• run_aware （bool，可选）：启用运行跟踪。违约： True
• run_timeout_seconds （float，可选）：关闭运行的非活动超时。违约： 30.0
• otlp_endpoint （str，可选）：OTLP收集器端点
• enable_console_export （bool，可选）：启用控制台跨度/度量输出
• logger （logging.Logger，可选）：自定义记录器实例

@observer.track(track_io=False)

装饰器为MCP工具添加可观察性。

参数：

• track_io （bool，可选）：启用完整的I/O跟踪（需要项目同意）。默认值：False

退货：

• 带有遥测功能的装饰异步功能

发展

运行测试

# Install development dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run with coverage
pytest --cov=mcp_observer --cov-report=html

代码质量

# Format code
black src tests

# Type checking
mypy src

# Linting
ruff check src tests

项目结构

mcp-observer-sdk/
├── src/
│   └── mcp_observer/
│       ├── __init__.py          # Package exports
│       ├── observer.py          # Main MCPObserver class
│       └── wrapper.py           # Decorator and telemetry logic
├── tests/
│   └── simple_example.py        # Example MCP server
├── pyproject.toml               # Package configuration
├── README.md                    # This file
└── LICENSE                      # MIT License

贡献

欢迎投稿！请随时提交拉取请求。对于重大更改，请先打开一个问题来讨论您想要更改的内容。

1. 分叉存储库
2. 创建功能分支(git checkout -b feature/AmazingFeature)
3. 提交您的更改(git commit -m 'Add some AmazingFeature')
4. 推到分支(git push origin feature/AmazingFeature)
5. 打开拉取请求

许可证

此项目根据MIT许可证获得许可-请参阅 许可证 文件以获取详细信息。

支持

• 问题:
• 文档: 全部文件
• 电子邮件:

致谢

• 为 模型上下文协议
• 由...驱动 开放遥测
• 受到对更好MCP工具可观察性需求的启发