彭博MCP服务器
引言
该项目提供了 模型上下文协议(MCP)服务器 与Bloomberg的Hypermedia API集成。\ 它允许客户端(例如Windsurf、Postman MCP、支持MCP集成的内部助理)获取 历史的 和 当前的 以结构化、标准化的格式进行费率计算。
服务器是用Python构建的,具有:
FastMCP用于MCP工具httpx用于异步HTTP请求- 服务器发送事件(SSE)来处理彭博社的异步数据分发
______________________________________________________________________
特性
- 🔑 基于JWT的身份验证 彭博超级媒体API
- 📈 历史汇率工具 –获取日期范围内的标准化时间序列数据
- 📊 当前汇率工具 –获取证券的最新/快照报价
- 🙋 问候/帮助工具 –带有使用示例的友好入口点
- ✅ 通过NocoDB API进行访问控制 –根据用户状态限制工具使用
- 🧩 架构验证 随着
msgspec - 📡 SSE调度员 –处理彭博社的异步响应
- 📝 结构化错误响应 用于一致的处理
- 📦 Docker化 随着
pyproject.toml+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_HOST_URL=https://api.bloomberg.com
BLOOMBERG_CLIENT_ID=
BLOOMBERG_CLIENT_SECRET=彭博配置:
BLOOMBERG_CATALOG_ID=
BLOOMBERG_API_VERSION=
BLOOMBERG_REQUEST_TIMEOUT=
BLOOMBERG_TERMINAL_USER_NUMBER=
BLOOMBERG_SECURITY_LIMIT=访问控制(NocoDB):
NOCODB_API_URL=https://nocodb.example.com/api/v1/db/data/v1/mydb/user_access
NOCODB_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 mainURL(可流式传输HTTP):
http://127.0.0.1:8000/mcp标头(可流式传输HTTP):
{
"Content-Type": "application/json",
"x-librechat-user-email": ""
}______________________________________________________________________
可用工具
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}
}
}
}______________________________________________________________________
身份验证和访问管理
彭博API要求 JWT令牌,使用客户端ID和密钥生成。
每个工具都实施访问控制:\ 服务器查询a NocoDB端点 使用登录的用户电子邮件(通过HTTP标头传递)。
- 用户必须存在并具有状态
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
- 日志记录:通过统一的控制台日志记录
src/utils/logging_config.py - 模式:请求/响应类型定义如下
msgspec - 错误处理:所有错误都使用共享
make_error_response助手 - 测试:包括以下示例
test_historical.py - 码头工人:轻量化
python:3.12-slim随着uv包管理器
______________________________________________________________________
