oig cloud MCP

oig_cloud_mcp 翻译为中文可以是“OIG云MCP”（假设OIG和MCP是特定项目或系统的缩写，且保持原样）。不过，具体翻译可能需要根据上下文或特定领域的术语来调整。如果OIG和MCP有特定的中文名称或含义，应相应地进行翻译

用于OIG云的官方MCP服务器 模型上下文协议Python SDK。

概述

这个服务器使用官方的FastMCP实现了模型上下文协议 mcp SDK。它为OIG Cloud提供包含会话管理和缓存功能的认证工具。

安装

pip install -r requirements.txt
pip install -e .

在运行任何Python命令行工具之前（例如 bin/cli_tester.py)，确保项目的虚拟环境已激活：

source .venv/bin/activate

用法

启动服务器

启动服务器：

python bin/main.py

或者使用提供的启动脚本：

./bin/start_server.sh

服务器将在 http://0.0.0.0:8000 使用可流式传输的HTTP传输方式。

你应该看到类似这样的输出：

SessionCache initialized.
Starting MCP Server with Authentication on http://0.0.0.0:8000
INFO:     Started server process [xxxxx]
INFO:     Waiting for application startup.
INFO:     StreamableHTTP session manager started
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

连接到服务器

MCP终端的访问地址为： http://localhost:8000/mcp

认证

服务器支持两种方法来验证您的OIG Cloud凭据。

1. 基本认证（推荐）

这是标准且最兼容的方法。您在（系统/应用中）提供您的凭证 Authorization 头球

如何生成令牌：

1. 输入你的电子邮箱和密码，并用一个冒号将它们连接起来： your_email@example.com:your_password
2. 将这个字符串进行Base64编码。

- 在Python中：

     import base64
     token = base64.b64encode(b'your_email@example.com:your_password').decode('utf-8')
     print(token)

- 在 Linux/macOS 命令行中：

     echo -n 'your_email@example.com:your_password' | base64

1. 最终的标题应该如下所示： Authorization: Basic

示例（包含） curl：

curl -X POST http://localhost:8000/mcp \
  -H "Authorization: Basic dGVzdEBleGFtcGxlLmNvbTp0ZXN0X3Bhc3N3b3Jk" \
  -H "Content-Type: application/json" \
  -d '{ ... tool call payload ... }'

有些客户只允许一个 Authorization: Bearer 头球 服务器将会 接受任一标签，前提是令牌是Base64编码的 email:password 成对的；一对。

示例使用 Bearer 使用curl标注：

curl -X POST http://localhost:8000/mcp \
  -H "Authorization: Bearer dGVzdEBleGFtcGxlLmNvbTp0ZXN0X3Bhc3N3b3Jk" \
  -H "Content-Type: application/json" \
  -d '{ ... tool call payload ... }'

2. 自定义头部（备用选项）

服务器还通过两个自定义头部接受凭据。这也是一种同样有效的（方法） 为偏好基于头部的凭证的客户端提供身份验证选项。

• X-OIG-Email您的OIG云邮箱。
• X-OIG-Password您的OIG云密码。

如果同时提供了基本认证和自定义头部信息，服务器将优先使用基本认证。

在Windows上进行Base64编码

如果你使用的是Windows系统，这里有几种将（数据）进行Base64编码的方法： email:password 字符串。

PowerShell / PowerShell Core (pwsh)：

[Convert]::ToBase64String([System.Text.Encoding]::UTF8.GetBytes('your_email@example.com:your_password'))

Windows PowerShell（旧版本直接使用.NET API）：

[System.Convert]::ToBase64String([System.Text.Encoding]::UTF8.GetBytes('your_email@example.com:your_password'))

使用 PowerShell 辅助工具的命令提示符（cmd.exe）（适用于大多数 Windows 电脑）：

@echo off
set "creds=your_email@example.com:your_password"
powershell -NoProfile -Command "[Convert]::ToBase64String([System.Text.Encoding]::UTF8.GetBytes('%creds%'))"

或者，在Windows Subsystem for Linux (WSL)、Git Bash或Cygwin上，你可以使用 标准的Linux命令：

echo -n 'your_email@example.com:your_password' | base64

使用Python MCP客户端

import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async def main():
    # Connect to the server
    async with streamablehttp_client("http://localhost:8000/mcp") as (read_stream, write_stream, _):
        async with ClientSession(read_stream, write_stream) as session:
            # Initialize the connection
            await session.initialize()
            print("✓ Connected")
            
            # List available tools
            tools = await session.list_tools()
            print(f"Available tools: {[tool.name for tool in tools.tools]}")
            
      # Call one of the OIG tools, for example get_basic_data
      result = await session.call_tool(
        "get_basic_data",
        arguments={
          "email": "user@example.com",
          "password": "your_password"
        }
      )
      if getattr(result, "structuredContent", None):
        import json
        print(json.dumps(result.structuredContent, indent=2))
      else:
        # Fallback to textual/stream content
        for content in getattr(result, "content", []):
          print(getattr(content, "text", content))

if __name__ == "__main__":
    asyncio.run(main())

提供了一个命令行测试工具 bin/cli_tester.py示例：

python bin/cli_tester.py get_basic_data
python bin/cli_tester.py get_extended_data --start-date 2025-01-01 --end-date 2025-01-31

使用Claude桌面版

添加到您的Claude桌面配置中（~/Library/Application Support/Claude/claude_desktop_config.json （在 macOS 上）：

{
  "mcpServers": {
    "oig-cloud": {
      "url": "http://localhost:8000/mcp"
    }
  }
}

使用curl进行手动测试

# Initialize connection
curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream, application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "initialize",
    "params": {
      "protocolVersion": "2024-11-05",
      "capabilities": {},
      "clientInfo": {"name": "test-client", "version": "1.0"}
    }
  }'

特点/特性

• FastMCP 集成使用官方MCP SDK的FastMCP，通过装饰器方式定义清晰的工具
• 会话管理缓存认证会话，具有可配置的驱逐时间（12小时）
• OIG工具： get_basic_data， get_extended_data，以及 get_notifications 提供了功能并从用户OIG Cloud账户中获取实时数据
• 可流式传输的HTTP传输现代、可扩展的基于HTTP的传输方式，支持SSE（服务器发送事件）

安全特性

此版本增加了对实时身份验证的基本安全保护：

• 白名单：仅限列表中的用户 whitelist.txt 被允许要求

认证并使用工具。添加您的OIG Cloud电子邮件（每行一个） to whitelist.txt 位于项目根目录。

• 速率限制：对每个用户的重复认证失败尝试进行追踪

并将采用指数退避策略暂时锁定账户 （默认设置：3次失败 -> 初始锁定10秒，每次加倍直至30秒）。

这些保护措施适用于单进程部署，并且 旨在成为一个简洁、易于理解的安全机制。用于生产 你应该将内存中的速率限制器替换为一个中央存储系统，如 Redis 的锁机制在跨进程和跨机器时仍然有效。

可用工具

get_basic_data 从已认证用户的OIG云账户获取光伏系统的实时快照，并从OIG云API返回实时数据负载。

get_extended_data - 获取指定时间段的历史时间序列数据。接受 start_date 和 end_date 参数（YYYY-MM-DD）将被转发到OIG Cloud API。

get_notifications - 从用户OIG云账户中获取系统警报、警告和信息消息。

建筑

oig_cloud_mcp/
├── bin/                    # Executable scripts
│   ├── main.py            # Server runner
│   ├── cli_tester.py      # Command-line test client
│   └── start_server.sh    # Startup script
├── src/oig_cloud_mcp/     # Python package
│   ├── __init__.py        # Package metadata
│   ├── tools.py           # MCP tool definitions
│   ├── session_manager.py # Session caching and API auth
│   ├── security.py        # Whitelist and rate limiting
│   └── transformer.py     # Data transformation utilities
├── tests/                 # Test suite
│   ├── fixtures/          # Test data
│   └── test_*.py          # Unit and integration tests
├── docs/                  # Documentation
├── requirements.txt       # Dependencies
└── setup.py              # Package configuration

配置

服务器设置可以在 bin/main.py:

oig_tools.settings.host = "0.0.0.0"  # Listen address
oig_tools.settings.port = 8000       # Listen port

会话缓存驱逐时间可以在 src/oig_cloud_mcp/session_manager.py:

session_cache = SessionCache(eviction_time_seconds=43200)  # 12 hours

可观测性

此服务器支持通过OpenTelemetry（OTel）实现全面的可观测性，并提供专门的日志记录功能用于安全监控，可与诸如（此处可添加具体工具名称，但原文未给出）等工具配合使用 fail2ban.

OpenTelemetry（追踪与日志）

要启用OTel，请配置以下环境变量：

• OTEL_EXPORTER_OTLP_ENDPOINT您的OTel收集器的gRPC或HTTP端点的完整URL（例如。， http://localhost:4317 用于 gRPC 或 http://localhost:4318/v1/logs （用于HTTP）。
• OTEL_EXPORTER_OTLP_PROTOCOL设置为 grpc （默认）或 http/protobuf 选择导出协议。
• OTEL_SERVICE_NAME这个服务的名称（默认为 oig-cloud-mcp）。

如果 OTEL_EXPORTER_OTLP_ENDPOINT 如果未设置，则OTel将被禁用。

Fail2ban的安全日志记录

服务器将所有失败的认证尝试记录到一个专用的日志文件中，该文件适合用于监控 fail2ban。

• 日志路径： 默认位置是 /var/log/oig_mcp_auth.log这可以通过设置来改变 FAIL2BAN_LOG_PATH 环境变量。
• 日志格式：

  YYYY-MM-DD HH:MM:SS: oig-mcp-auth: FAILED for user [user@email.com] from IP [123.45.67.89]

在Docker中运行时，您应该将一个卷挂载到此路径，以便在主机上持久化保存日志。

发展

服务器使用：

• FastMCP带有装饰器的高级MCP服务器框架
• StreamableHTTP 可以翻译为“可流式传输的HTTP”或“支持流式处理的HTTP”，具体取决于上下文和使用场景。在这里，我选择了一个较为通用的翻译：“可流式传输的HTTP”基于现代HTTP的MCP传输方案
• 会话缓存12小时会话缓存以减少身份验证调用
• Uvicorn（通常指的是一个用于构建ASGI服务器的Python库，可直接用于运行ASGI应用，如FastAPI等）适用于生产环境部署的ASGI服务器

测试

这个项目包含一个全面的测试套件，其中包括单元测试、集成测试以及自动化的持续集成/持续部署（CI/CD）流水线。

快速入门

# Install development dependencies
pip install -r requirements-dev.txt

# Run all tests
pytest

# Run tests with verbose output
pytest -v

# Check code quality
flake8 .
black --check .

如需详细的测试文档，请参阅 \docs/testing.md\ 翻译成中文是：“文档/测试.md”.

测试覆盖率

• 单元测试纯函数在 transformer.py 和 security.py
• 集成测试工具端点，使用模拟API调用
• CI/CD（持续集成/持续交付）在GitHub Actions上为Python 3.13进行自动化测试

故障排除

404 页面未找到

确保你连接的是 /mcp 终端点；端点 http://localhost:8000/mcp

406 不可接受

确保您的客户端发送正确的Accept头部信息：

• Accept: text/event-stream, application/json

连接被拒绝

验证服务器是否正在运行：

curl http://localhost:8000/mcp

你应该能看到一个响应（即使它是关于头部的错误）。