Market Macro Agent

​

Architecture

Browser (vanilla JS)
  │
  ├── GET  /stream/state  (SSE — pipeline tiles, activity feed, row counts)
  │
  └── POST /chat          (SSE — tool calls, partial text, answers)
           │
           └── agent/loop.py
                 │
                 ├── Anthropic API (Claude)
                 │     ↕ tool_use / end_turn
                 │
                 ├── agent/executor.py
                 │     └── MCP tools → Datris MCP Server (port 3000)
                 │           (run_tap, get_pipeline_status, query_postgres,
                 │            list_taps, get_tap_logs, …)
                 │
                 └── agent/scheduler.py → run_tap every N minutes

              ┌──────────────────────────────────────┐
              │ Datris Platform                      │
              │  Tap runner (Docker sandbox)         │
              │    fred_tap → FRED API               │
              │    equities_tap → yfinance           │
              │    crypto_tap → CoinGecko            │
              │    sec_tap → SEC EDGAR               │
              │           ▼                          │
              │   Pipelines (Postgres destinations)  │
              └──────────────────────────────────────┘

​

How It Works

1. On startup — Connects to the Datris MCP server via SSE, discovers tools via tools/list, loads the server’s instructions and resources, pushes FRED_API_KEY from .env into Datris as a tap secret (create_tap_secret), and provisions four taps (fred_tap, equities_tap, crypto_tap, sec_tap), each wired to its own pipeline. Idempotent: re-running against an existing install is a no-op. Taps are created with scheduling disabled so the agent’s own loop drives refreshes.
2. User asks a question — MacroAgent determines which data family is relevant
3. Refresh — The agent calls run_tap. The platform runs the tap’s Python fetch() in a Docker sandbox and submits records into the pipeline
4. Wait for ingestion — Ingestion is asynchronous; the agent polls get_pipeline_status with the run’s publisherToken until every row’s state is end or error
5. Query — The agent queries the landed data via query_postgres and answers grounded in actual numbers
6. Background refresh — Every REFRESH_INTERVAL_MINUTES the scheduler calls run_tap for every pipeline that has been exercised at least once
7. Tap health — Ask about failures and the agent calls get_tap_logs to report the last runs and any errors

​

Data Sources

​

Quick Start

cd examples/market-macro-agent
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env
# Edit .env → add ANTHROPIC_API_KEY (and optionally FRED_API_KEY)

# Make sure Datris is running:
cd ../../ && docker compose up -d && cd examples/market-macro-agent

uvicorn main:app --reload --port 8001

​

Demo Queries

• "What's the current macro picture?" — runs the FRED + equities taps, then queries
• "Refresh all taps" — triggers run_tap across all four data families
• "Is crypto confirming the risk-on trade in equities?" — cross-pipeline analysis
• "Show me the last few tap runs" — exercises get_tap_logs live

​

Environment Variables

​

Requirements

• Python 3.11+
• Datris Platform running via Docker
• Anthropic API key