North MCP Python SDK

North MCP Python SDK

此 SDK 基于原始 SDK 构建。一般信息请参阅原始仓库的 README。本文档重点介绍 North 特有的细节。

安装

uv pip install git+ssh://[email protected]/cohere-ai/north-mcp-python-sdk.git

为什么要创建此仓库

此仓库提供相关代码，让你的服务可以使用 North 进行身份验证。North 是对原始规范的自定义扩展。除此之外，SDK 没有任何改动；此 SDK 是在原始 SDK 的基础上构建的。

主要差异

• North 仅支持 StreamableHTTP 传输。sse 传输已弃用，虽然为了向后兼容仍可使用，但创建新服务时不应使用它
• 你可以使用 ID 令牌验证 North 用户身份，并可在生产环境中通过 trusted_issuers 限制接受的身份提供方。
• 你可以访问用户的 OAuth 令牌，代表用户与第三方服务交互。
• 你可以访问用户的身份信息（来自与 North 配合使用的身份提供方）。
• 调试模式：提供详细的身份验证日志和问题排查信息。
• OpenTelemetry 辅助工具：支持自定义跨度、日志与追踪关联，以及保护隐私的错误记录（基于 FastMCP 遥测）。
• 内置健康检查端点，可用于 Kubernetes 存活探针（默认启用）。

健康检查

NorthMCPServer 内置了 /health 端点，可响应 GET 请求，并返回 200 OK 纯文本响应。该端点适用于 Kubernetes 存活／就绪探针以及负载均衡器健康检查。此端点不经过身份验证，因此无需令牌。

此功能默认启用。要禁用它：

mcp = NorthMCPServer(name="Demo", health_check=False)

示例

此仓库包含可用作快速入门的示例服务。你可以在示例目录中找到它们。

示例涵盖身份验证、North UI 的工具元数据、调试模式和 OpenTelemetry（examples/telemetry-demo/）。

身份验证

此 SDK 提供多种策略，用于验证用户身份并授权其请求。

我只希望 North 能向我的服务发送请求

North 会在请求中发送 X-North-ID-Token。在生产环境中，请配置 trusted_issuers，以便服务验证该令牌是否由预期的身份提供方签发。对于本地开发或风险较低的场景，此设置为可选；令牌会在不验证签名的情况下进行解码。

mcp = NorthMCPServer(
    name="Demo",
    port=5222,
    trusted_issuers=["https://your-idp.example.com"],
)

我想获取调用服务的 North 用户身份

请参阅 examples/server_with_auth.py。在处理请求时，调用以下内容：

user = get_authenticated_user()
print(user.email)

我需要通过 OAuth 访问第三方服务（例如 Google Drive、Slack 等）

与上面类似：

user = get_authenticated_user()
print(user.connector_access_tokens)

调试模式

North MCP SDK 包含全面的调试模式，可详细记录身份验证过程、传入请求和令牌验证信息。排查身份验证问题时，这些信息非常有用。

启用调试模式

启用调试模式有多种方式：

1. 环境变量（推荐）

export DEBUG=true
python your_server.py

2. 构造函数参数

mcp = NorthMCPServer(name="Demo", port=5222, debug=True)

调试模式会记录哪些内容

启用调试模式后，你会看到详细日志，包括：

• 请求标头：所有传入的 HTTP 标头（包括 Authorization）
• 令牌解析：身份验证令牌的 Base64 解码和 JSON 解析
• JWT 验证：用户 ID 令牌的解码和验证步骤
• 身份验证详情：用户邮箱、可用连接器、令牌数量
• 错误上下文：包含问题排查上下文的详细错误消息

调试输出示例

2024-01-15 10:30:45 - NorthMCP.Auth - DEBUG - Authenticating request from ('127.0.0.1', 54321)
2024-01-15 10:30:45 - NorthMCP.Auth - DEBUG - Request headers: {'authorization': 'Bearer eyJ...', 'content-type': 'application/json'}
2024-01-15 10:30:45 - NorthMCP.Auth - DEBUG - Authorization header present (length: 248)
2024-01-15 10:30:45 - NorthMCP.Auth - DEBUG - Successfully decoded base64 auth header
2024-01-15 10:30:45 - NorthMCP.Auth - DEBUG - Successfully parsed auth tokens. Has user_id_token: True, Connector count: 2
2024-01-15 10:30:45 - NorthMCP.Auth - DEBUG - Available connectors: ['google', 'slack']
2024-01-15 10:30:45 - NorthMCP.Auth - DEBUG - Successfully decoded user ID token. Email: [email protected]
2024-01-15 10:30:45 - NorthMCP.AuthContext - DEBUG - Setting authenticated user in context: [email protected], connectors=['google', 'slack']

调试模式示例

请参阅 examples/server_with_debug.py 获取可运行的示例。

安全提示

调试模式会记录敏感信息，包括请求标头和令牌元数据。切勿在生产环境中启用调试模式，否则日志可能会暴露身份验证信息。

OpenTelemetry

配置 TracerProvider 后，FastMCP 3 会为每次工具调用生成一个 span。North SDK 在此基础上添加了辅助工具；它不会安装导出器或配置跟踪数据的发送位置——opentelemetry-sdk 和 OTLP 导出器不是捆绑的依赖项。即使没有 TracerProvider，服务也能正常运行；在添加 TracerProvider 之前，跟踪数据不会执行任何操作。

SDK 提供的功能

辅助工具	用途
TelemetryConfig	遥测集成的可选配置对象（敏感数据策略、日志/跟踪关联）
mcp.telemetry.traced_span	在 FastMCP 工具 span 下嵌套自定义 span，由服务的 TelemetryConfig 驱动。适用于可访问 mcp 的场景。
traced_span (模块级)	相同的上下文管理器，但会通过 fastmcp.server.dependencies.get_server 自动解析活动服务的配置。适用于不持有 mcp 引用的工具函数体。
get_telemetry_config()	返回活动 NorthMCPServer 的 TelemetryConfig（若没有活动的 North 服务，则返回私有的“关闭”配置）。即使没有 mcp 引用，也可让工具根据 record_sensitive_data 执行分支逻辑。
Depends	FastMCP 的 Depends 工厂的重新导出项。与 get_telemetry_config 搭配使用，可实现 FastAPI 风格的参数注入：telemetry: TelemetryConfig = Depends(get_telemetry_config)。
get_tracer	FastMCP 的 tracer（fastmcp 插桩名称）的重新导出项
TraceContextFormatter	活动 span 存在时，将 trace_id / span_id 附加到 NorthMCP.{name} 记录器的日志行中

设置

1. 在 main 中注册 TracerProvider 和导出器，然后再导入 NorthMCPServer（FastMCP 会在导入时读取全局提供程序）。可使用自己的设置，或使用标准 OTel 环境变量（OTEL_EXPORTER_OTLP_ENDPOINT、OTEL_SERVICE_NAME 等）。
2. 将 TelemetryConfig 传递给 NorthMCPServer——这是选择启用的信号。启用 log_trace_context 后（选择启用的配置默认启用此项），跟踪/跨度 ID 会自动附加到 NorthMCP.* 日志行中。
3. 使用 Depends(get_telemetry_config) 将配置注入工具，并调用 telemetry.traced_span(...) 创建自定义 span。有关 mcp.telemetry 和直接导入替代方案，请参阅辅助工具表。

def configure_tracing() -> None:
    # Install opentelemetry-sdk + exporters in your server project.
    # See examples/telemetry-demo/main.py for a minimal OTLP gRPC example.
    ...

def main() -> None:
    configure_tracing()

    from north_mcp_python_sdk import (
        Depends,
        NorthMCPServer,
        TelemetryConfig,
        get_telemetry_config,
    )

    mcp = NorthMCPServer(
        "Search",
        telemetry=TelemetryConfig(
            record_sensitive_data=False,
            log_trace_context=True,
        ),
    )

    @mcp.tool()
    async def search(
        query: str,
        telemetry: TelemetryConfig = Depends(get_telemetry_config),
    ) -> list[dict]:
        # FastMCP injects the active server's TelemetryConfig as `telemetry`
        # before the tool body runs. No reference to `mcp` is needed, and
        # `telemetry.traced_span` honours `record_sensitive_data` for
        # exception details.
        with telemetry.traced_span(
            "search.provider_call",
            attributes={"search.query_length": len(query)},
        ) as span:
            if telemetry.record_sensitive_data:
                span.add_event("search.query", {"query": query})

            results = await provider.search(query)
            span.set_attribute("search.result_count", len(results))
            return results

在默认值中使用 Depends(...) 会触发 ruff 的 B008 和 basedpyright 的 reportCallInDefaultInitializer。请按照此仓库的 pyproject.toml 配置，将 Depends 加入白名单；FastAPI 项目也采用相同做法。

完整演练：examples/telemetry-demo/main.py。如需更广泛的自动插桩（HTTP、日志等），请按照 FastMCP 遥测文档 使用 opentelemetry-instrument CLI。

TelemetryConfig 选项

选项	默认值（选择启用时）	说明
record_sensitive_data	False	设为 True 时，traced_span 会在 span 中记录完整的异常消息，工具也可将敏感载荷（查询文本、ID 等）作为 span 事件附加。为安全起见，默认关闭；只能在代码中明确选择启用。
log_trace_context	True	设为 True 时，SDK 会附加一个格式化器，在活动 span 存续期间，将 trace_id / span_id 附加到 NorthMCP.* 记录器的日志行中。

如果将 telemetry=None，SDK 会使用内部的“未选择启用遥测”配置：两个标志均为 False，不会为 NorthMCP.* 记录器附加格式化器，并且 traced_span 发生错误时只记录异常类型名称。这两个标志均没有环境变量回退值。只要配置了 TracerProvider，FastMCP 自身的工具/资源/提示词 span 不受影响——它们仍会显示。

mcp = NorthMCPServer(
    name="Demo",
    telemetry=TelemetryConfig(record_sensitive_data=True),
)

不使用 North 进行本地开发

本指南介绍如何在不连接 North 的情况下，在本地测试 MCP 服务。为此，我们将使用 MCP Inspector。你可以通过以下命令运行它：

npx @modelcontextprotocol/inspector

如果不需要身份验证，只想在本地运行，可以选择 stdio 传输方式。打开 MCP Inspector，并按如下方式配置：

• 传输类型：stdio
• 命令：uv
• 参数：run examples/server_with_auth.py --transport stdio

接下来：

• 点击“Connect”
• 选择屏幕顶部的“Tools”。
• 点击“List Tools” -> “add”
• 输入数字并点击“Run”。你应该能看到它们的和。

添加身份验证

如果想在本地测试身份验证机制，可以按以下步骤操作。首先，使用 streamable http 传输方式启动服务：

uv run examples/server_with_auth.py --transport streamable-http

接下来，创建一个 bearer token。你可以使用 examples/create_bearer_token.py 生成一个，也可以使用预先生成的 token。

打开 MCP Inspector，并按如下方式配置：

• 传输类型：Streamable HTTP
• URL：http://localhost:5222/mcp
• Authentication -> Bearer token：eyJzZXJ2ZXJfc2VjcmV0IjogInNlcnZlcl9zZWNyZXQiLCAidXNlcl9pZF90b2tlbiI6ICJleUpoYkdjaU9pSklVekkxTmlJc0luUjVjQ0k2SWtwWFZDSjkuZXlKbGJXRnBiQ0k2SW5SbGMzUkFZMjl0Y0dGdWVTNWpiMjBpZlEuV0pjckVUUi1MZnFtX2xrdE9vdjd0Q1ktTmZYR2JuYTVUMjhaeFhTaEZ4SSIsICJjb25uZWN0b3JfYWNjZXNzX3Rva2VucyI6IHsiZ29vZ2xlIjogImFiYyJ9fQ==

按照之前相同的步骤操作。调用工具时，你应该能在启动服务的终端中看到以下日志：

This tool was called by: [email protected]

开发

先决条件

要为此项目做出贡献，你需要：

• Python 3.11+：SDK 所需
• uv >= 0.8.13：用于依赖管理、代码格式化和 CI 检查

设置

1. 克隆代码库：

   git clone https://github.com/cohere-ai/north-mcp-python-sdk.git
   cd north-mcp-python-sdk
   

2. 安装依赖：

   uv sync --dev
   

代码格式化

此项目使用 uv format 统一代码格式。CI 流水线会强制执行这些规范：

# Check formatting (same as CI)
uv format --preview-features format --check

# Apply formatting
uv format --preview-features format

运行测试

uv run pytest

贡献

提交 PR 前：

1. 确保代码通过格式检查：uv format --preview-features format --check
2. 运行测试套件：uv run pytest
3. 遵循现有的代码风格和模式