Bloomberg Mcp Server

Bloomberg MCP Server

はじめに

このプロジェクトは、Bloomberg の Hypermedia API と連携する Model Context Protocol（MCP）サーバーを提供します。
クライアント（Windsurf、Postman MCP、MCP 連携に対応した社内アシスタントなど）は、過去および現在のレートを構造化された正規化済みの形式で取得できます。

このサーバーは Python で構築されており、以下を使用します。

• MCP ツール向けの FastMCP
• 非同期 HTTP リクエスト向けの httpx
• Bloomberg の非同期データ配信を処理する Server-Sent Events（SSE）

---

機能

• 🔑 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：uv パッケージマネージャーを使用した軽量な python:3.12-slim

---