Bloomberg Mcp Server

Bloomberg MCP 服务

简介

本项目提供一个与 Bloomberg Hypermedia API 集成的 Model Context Protocol (MCP) 服务。
它允许客户端（例如 Windsurf、Postman MCP、支持 MCP 集成的内部助手）以结构化、规范化的格式获取历史和当前汇率。

该服务使用 Python 构建，采用：

• FastMCP用于 MCP 工具
• httpx用于异步 HTTP 请求
• Server-Sent Events (SSE)，用于处理 Bloomberg 的异步数据分发

---

功能

• 🔑 使用 Bloomberg Hypermedia API 进行 基于 JWT 的身份验证
• 📈 历史汇率工具 – 获取指定日期范围内的规范化时间序列数据
• 📊 当前汇率工具 – 获取证券的最新/快照报价
• 🙋 问候/帮助工具 – 提供友好的入口和用法示例
• ✅ 通过 NocoDB API 进行访问控制 – 根据用户状态限制工具使用
• 🧩 使用 msgspec进行 架构验证
• 📡 SSE 分发器 – 处理 Bloomberg 的异步响应
• 📝 结构化错误响应，确保处理方式一致
• 📦 使用 pyproject.toml + uv 进行依赖管理，并支持 Docker 容器化
• 🖥️ 仅输出到控制台的日志记录，支持配置日志级别

---

安装

克隆代码仓库并使用 uv 安装依赖：

git clone https://github.com/AlterDomus/bloomberg-mcp-server.git
cd bloomberg-mcp-server

# install dependencies (pyproject.toml)
uv venv # Optional, if you want to create a virtual enviornment (recommended)
uv sync

---

配置

所有配置均通过环境变量完成。

Bloomberg 凭据：

BLOOMBERG_HOST_URL=https://api.bloomberg.com
BLOOMBERG_CLIENT_ID=<your client id>
BLOOMBERG_CLIENT_SECRET=<your client secret>

Bloomberg 配置：

BLOOMBERG_CATALOG_ID=<catalog id>
BLOOMBERG_API_VERSION=<version number>
BLOOMBERG_REQUEST_TIMEOUT=<request timeout>
BLOOMBERG_TERMINAL_USER_NUMBER=<users terminal number>
BLOOMBERG_SECURITY_LIMIT=<default security limit>

访问控制（NocoDB）：

NOCODB_API_URL=https://nocodb.example.com/api/v1/db/data/v1/mydb/user_access
NOCODB_TOKEN=<nocodb-api-token>

日志记录：

LOG_LEVEL=INFO  # DEBUG, INFO, WARNING, ERROR

---

运行服务

本地（stdio）：

python -m main

路径（stdio）：

C:\Projects\bloomberg-mcp\.venv\Scripts\python.exe C:\Projects\bloomberg-mcp\main.py # Path consist of env and the main file location

本地（可流式传输的 HTTP）：

python -m main

URL（可流式传输的 HTTP）：

http://127.0.0.1:8000/mcp

标头（可流式传输的 HTTP）：

{
  "Content-Type": "application/json",
  "x-librechat-user-email": "<logged-in user from MCP client>"
}

---

可用工具

greeting

用途：返回友好的介绍和示例。
输入：可选负载。
输出：

{
  "message": "👋 Hi! I’m your Bloomberg AI Assistant..."
}

---

historical_rates

用途：检索证券的规范化历史时间序列。
输入：

{
  "symbols": [{"id_type":"TICKER","id_value":"EURUSD Curncy"}],
  "fields": [{"name":"PX_LAST"},{"name":"PX_MID"}],
  "currency": "USD",
  "start_date": "2024-01-01",
  "end_date": "2024-01-05"
}

输出：

{
  "data": {
    "TICKER:EURUSD Curncy": [
      {"date":"2024-01-01","values":{"PX_LAST":1.23,"PX_MID":1.22}}
    ]
  }
}

---

current_rates

用途：检索最新/快照报价。
输入：

{
  "symbols": [{"id_type":"SEDOL","id_value":"2046251"}],
  "fields": [{"name":"BID"},{"name":"ASK"},{"name":"PX_LAST"}],
  "currency": "EUR"
}

输出：

{
  "data": {
    "SEDOL:2046251": {
      "fields": {"BID":189.7,"ASK":190.3,"PX_LAST":190.1}
    }
  }
}

---

身份验证和访问管理

Bloomberg API 需要使用客户端 ID 和密钥生成 JWT 令牌。

每个工具都会强制执行访问控制：
服务会使用已登录用户的邮箱（通过 HTTP 标头传入）查询 NocoDB 端点。

• 用户必须存在且状态为 active
• 否则，工具会返回结构化的 ACCESS_DENIED 错误。

---

开发说明

项目布局：

scripts/
  sample_current.json
  sample_historical.json
  test_historical.py
src/
  config.py
  server.py
  schemas/
    base.py
    current.py
    error.py
    historical.py
  services/
    sse.py
    access.py
  tools/
    greeting.py
    historical.py
    current.py
    _guards.py
  utils/
    auth.py
    errors.py
    http.py
    logging_config.py
    parsing.py
main.py
.env
.env.example
<configuration files>

• 日志记录：通过 src/utils/logging_config.py 统一记录控制台日志
• 架构：使用 msgspec 定义请求/响应类型
• 错误处理：所有错误均使用共享的 make_error_response 辅助函数
• 测试：包含 test_historical.py 等示例
• Docker：使用轻量级的 python:3.12-slim 镜像和 uv 包管理器

---