Horizon MCP Demo -- Extended v2

Companion reference implementation for the Horizon Data Partners white paper The Governed Data Layer: Why AI Agents Fail Without One, and How to Build It.

Two governed systems, one agent:

                         +---------------------------+
                         |       Claude agent        |
                         | scripts/run_two_server_   |
                         |        agent.py           |
                         +------+-------------+------+
                                |  MCP (SSE)  |
                 +--------------+             +---------------+
                 v                                            v
  +-----------------------------+            +--------------------------------+
  |  Pipeline Semantic Layer    |            |  Predictive Model Service      |
  |  port 8001                  |            |  port 8002                     |
  |  dbt Core + DuckDB          |            |  DuckDB segment store          |
  |  list_models                |            |  describe_inputs               |
  |  get_model_details          |            |  describe_outputs              |
  |  get_metric_definitions     |            |  validate_payload              |
  |  query_data                 |            |  score_opportunity             |
  |  owns: win_rate, fees,      |            |  owns: segments, matching,     |
  |  margin, velocity           |            |  the 6 outcome estimates       |
  +-----------------------------+            +--------------------------------+

Why v2 exists

The first extension (horizon-mcp-demo-extended) wired a P&C insurance semantic layer into this same predictive model. Every tool call succeeded and the agent produced numerically precise, analytically meaningless answers -- insurance policies are not professional services opportunities. The failure was not technical; it was the absence of cross-system governance.

v2 fixes both failures of v1:

1. Governance first. docs/CROSS_SYSTEM_GOVERNANCE.md was written and approved before any code: a domain compatibility check, an approved exact field mapping (no assumptions -- and the document explains why that is the correct baseline), and a governed interface definition that the model server returns from describe_inputs and enforces before scoring. The server refuses out-of-contract input; it does not adapt it.

2. SSE transport, not stdio. Both MCP servers run as independent local HTTP processes (ports 8001/8002) and the agent connects with mcp.client.sse.sse_client. With no stdio subprocesses nested inside the agent's event loop, the Windows anyio BrokenResourceError from v1 cannot occur, and the fix generalizes to any number of servers.

Related MCP server: career-scout-mcp

Repository layout

docs/CROSS_SYSTEM_GOVERNANCE.md    governance doc (read this first)
seeds/                             opportunities.csv, jobs.csv, pipeline_summary.csv
models/staging, models/marts       dbt models (the semantic layer)
models/schema.yml                  descriptions + tests served through MCP
model_data/analytical_segments.csv the model's trained segment artifact
model_store/                       built by scripts/build_model_store.py (gitignored)
mcp_server/pipeline_mcp_server.py  System A, port 8001 (SSE)
mcp_server/predictive_model_mcp_server.py  System B, port 8002 (SSE)
scripts/build_model_store.py       loads the segment CSV into DuckDB
scripts/smoke_test_servers.py      exercises all 8 tools, no API key needed
scripts/run_two_server_agent.py    the 5-question agent demo

Quickstart (Windows PowerShell)

From the repository root:

# 1. environment
python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install -r requirements.txt

# 2. build the semantic layer (profiles.yml is at the repo root; run from here)
dbt seed
dbt run
dbt test

# 3. build the model's segment store
python scripts\build_model_store.py

# 4. start the servers -- one PowerShell terminal each
python mcp_server\pipeline_mcp_server.py           # terminal 1, port 8001
python mcp_server\predictive_model_mcp_server.py   # terminal 2, port 8002

# 5. pre-flight (terminal 3) -- 17 checks, no API key required
python scripts\smoke_test_servers.py

# 6. run the agent demo (terminal 3)
$env:ANTHROPIC_API_KEY = "sk-ant-..."
python scripts\run_two_server_agent.py

The five demo questions

1. What does the pipeline semantic layer contain and what metrics are governed?

2. What are the win rates and average fees by service line and client type in the pipeline data?

3. Score a new Advisory/Business/Healthcare/New/Referral opportunity -- what are the six outcome estimates and how reliable is the segment match? (Deliberately chosen: its finest segment has only 27 observations, under the 30 minimum, so outcomes match at levels 1-2 and the response says so.)

4. For open opportunities in the pipeline that have been open longer than the model's estimated DaysSellToStart, identify which ones are delayed and by how much. (Cross-system: mart_open_pipeline from System A, estimates per distinct segment from System B, joined by the agent.)

5. Which service line has the highest fee-to-margin ratio in the historical data, and what does the model predict for win probability in that segment going forward? (Historical metric and forward estimate side by side, from independent definitions.)

Token logging

Each run writes logs/agent_run_<timestamp>.json in the same format as the original horizon-mcp-demo, extended per server:

• every tool_calls entry carries a server field;

• question_totals and run_totals carry by_server, holding tool_call_count and attributed API tokens. Attribution rule: an API call whose response invokes tools has its tokens split evenly across the distinct servers invoked in that response; synthesis calls (no tools) are attributed to "synthesis".

Data notes

• The pipeline seeds are a static snapshot: days_open for open opportunities is as of 2026-07-15, so demo output is reproducible.

• model_data/analytical_segments.csv is the model's trained artifact (the DuckDB analogue of dbo.analytical_segments from HorizonDataPredictiveModel), built from a synthetic 1,400-record training corpus. Every coarser level is the exact observation-weighted aggregate of its children, so the numbers survive scrutiny.

• Dimension drop order when coarsening: LeadSource, NewVsExisting, Industry, ClientType, ServiceLine (level 0 finest to level 5 overall) -- the same order as usp_BuildSegments.

Troubleshooting

• Port already in use -- something else holds 8001/8002. Find it with netstat -ano | findstr :8001, stop it, or change the PORT constant at the top of the affected server file (and the URL in both scripts).

• Windows Firewall prompt on first server start -- uvicorn binding 127.0.0.1 can trigger a Defender prompt. Allow access; the servers bind loopback only.

• DuckDB lock errors from dbt -- stop the pipeline server before re-running dbt seed/dbt run; the server holds a read-only connection but dbt needs the writer.

• manifest.json not found / Model store not found -- run steps 2 and 3 of the quickstart before starting the servers.

• Agent cannot connect -- both servers must be running first; the agent does not spawn them (that is the point of the SSE architecture).

Relationship to the other repositories

• horizon-mcp-demo -- the original single-server insurance demo. Untouched by this project.

• HorizonDataPredictiveModel -- the SQL Server implementation this model service ports. The DuckDB segment-matching SQL mirrors usp_ApplyModel's CROSS APPLY pattern: per-outcome, OR-IS-NULL per dimension, finest qualifying level, minimum 30 observations.