Newsroom Evidence Finder

Introduction

30 Second Summary

Public records often give the same company several names. A reporter still needs to trace every promising lead back to the record that produced it.

In this project, you will build a local DuckDB evidence finder that turns UK public contracting records into traceable reporting leads. The finished workflow uses Claude for source-backed entity extraction that you can inspect through MCP Inspector.

What You'll Build

Picture a supplier search in MCP Inspector returning a reporting lead beside the public source record that supports it.

By the end of this project, you'll have:

• An entity resolution demo where an exact-name join returns zero. The normalised pipeline then links the supplier to its company record.
• A traceable Claude extraction that keeps every person or company beside an exact source quote. Each result also carries its source record and model ID.
• An interview-ready GitHub demo with deterministic tests that run without live API calls. Two MCP tools let you search records or trace an entity to its evidence.
• Secret Mission: Design and test an editorial evidence-quality gate that flags records a reporter should not trust yet.

Are there any prerequisites?

This project requires a Windows computer with internet access. A GitHub account plus basic Python knowledge completes the prerequisites.

Before We Start

This is your moment to commit to an evidence-first reporting tool before the hands-on work begins. Every surfaced entity must remain traceable to its source record, exact quote, and provenance reference.

Set Up the Toolchain and Protect Secrets

A reproducible evidence finder needs the same dependencies every time it runs. It also needs a clean Git history that never captures an API key.

You will prepare Visual Studio Code, PowerShell, Node.js, Python, and uv. You will then create the repository with pinned packages plus protected local credentials.

In this step, get ready to:

• Install or verify every tool required by the local project.
• Create the repository with pinned dependencies plus starter files.
• Create the API credentials without exposing them to Git.

Install and verify the Windows tools

Each tool has a specific role in the workflow. Visual Studio Code holds the project files. Git tracks safe changes. Node.js supports MCP Inspector. Python plus uv run the evidence pipeline.

• Press the Windows key to open Windows search.
• Type Visual Studio Code into the search field.

✔️ Visual Studio Code opens

• Select Visual Studio Code from the search results.

Your editor is ready to hold the project workspace.

ⓧ Visual Studio Code is not installed

• Visit the official Visual Studio Code Windows setup page.
• Download the User setup installer.
• Run the installer with the recommended options.
• Select Visual Studio Code from Windows search after installation.

The User setup installs without administrator permissions. It also adds the Visual Studio Code command to your system path.

• Press the Windows key to open Windows search.
• Type PowerShell into the search field.
• Press Enter to open PowerShell.
• Check the installed Git version by running:

git --version

What does this check do?

The command asks Git for its installed version. The result shows whether the required Windows build is available.

✔️ I see the required Git version

Good progress. Git for Windows 2.56.0(2) is ready to create the repository.

ⓧ I see an older Git version

• Visit the official Git for Windows installation page.
• Download the maintained x64 build 2.56.0(2).
• Run the installer with its default options.
• Restart PowerShell after installation.
• Repeat the Git version check above.

ⓧ Git is not recognized

• Visit the official Git for Windows installation page.
• Download the maintained x64 build 2.56.0(2).
• Run the installer with its default options.
• Restart PowerShell so Git is available on the system path.
• Repeat the Git version check above.

• Check the installed Node.js version by running:

node --version

Why check Node.js?

MCP Inspector depends on Node.js through npx. The version check confirms that the project can launch Inspector later.

✔️ I see the required Node.js version

Node.js v24.22.0 LTS is ready for the MCP Inspector workflow.

ⓧ I see an older Node.js version

• Visit the official Node.js download page.
• Download Node.js v24.22.0 LTS for Windows.
• Run the downloaded installer.
• Restart PowerShell after installation.
• Repeat the Node.js version check above.

ⓧ Node.js is not recognized

• Visit the official Node.js download page.
• Download Node.js v24.22.0 LTS for Windows.
• Run the downloaded installer.
• Restart PowerShell so Node.js is available on the system path.
• Repeat the Node.js version check above.

• Check the installed uv version by running:

uv --version

Why use uv?

uv creates the local Python environment from the versions in pyproject.toml. This keeps package versions consistent across machines.

✔️ I see the pinned uv release

uv v0.13.1 is ready to manage Python plus the project packages.

ⓧ I see an older uv version

• Install uv v0.13.1 by running the official Windows installer command:

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/0.13.1/install.ps1 | iex"

What does this command do?

PowerShell downloads the pinned uv installation script from the official host. It then installs that release for your user account.

• Restart PowerShell after installation.
• Repeat the uv version check above.

ⓧ uv is not recognized

• Install uv v0.13.1 by running the official Windows installer command:

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/0.13.1/install.ps1 | iex"

What does this command do?

PowerShell downloads the pinned uv installation script from the official host. It then installs uv for your user account.

• Restart PowerShell so uv is available on the system path.
• Repeat the uv version check above.

• Check the installed Python version by running:

python --version

Why Python 3.11 or Later?

The MCP Python SDK requires Python 3.11 or later. uv can install the required interpreter when the system version is too old.

✔️ I see Python 3.11 or later

Your Python interpreter meets the project requirement. uv can use it when the environment is synchronized.

ⓧ I see an older Python version

• Install Python 3.11 through uv by running:

uv python install 3.11

What does this command do?

uv downloads a managed Python 3.11 interpreter. You will select it for the repository after the project folder exists.

ⓧ Python is not recognized

• Install Python 3.11 through uv by running:

uv python install 3.11

What does this command do?

uv downloads a managed Python 3.11 interpreter. The repository can use this interpreter without depending on a system Python installation.

Still seeing an old version?

Close every PowerShell window after an installation. Open a fresh PowerShell window so Windows reloads the system path.

If the old result remains, help me diagnose a Windows tool version problem.

Create the repository and pinned environment

The repository needs a predictable location plus a fixed dependency list. The starter files also reserve clear places for the pipeline code, MCP adapter, tests, documentation, secret template, and personal notes.

• Create the newsroom-evidence-finder folder on your Desktop by running:

New-Item -Path "$env:USERPROFILE\Desktop" -Name "newsroom-evidence-finder" -ItemType "Directory"

What does this command do?

PowerShell creates the project folder directly on your Desktop. Its output lists newsroom-evidence-finder as the new directory.

• Move PowerShell into the new project folder by running:

Set-Location -Path "$env:USERPROFILE\Desktop\newsroom-evidence-finder"

What does this command do?

PowerShell now treats newsroom-evidence-finder as the current location. Every project command that follows applies to this folder.

• Initialize the repository on the main branch by running:

git init -b main

What does this command do?

Git creates its local repository metadata. The -b main option names the first branch main.

The initialized-repository message confirms that source control is active inside the correct folder.

• Open the current project folder in Visual Studio Code by running:

code .

What does this command do?

The dot represents the current PowerShell folder. Visual Studio Code opens newsroom-evidence-finder as one workspace.

• Confirm the workspace is trusted if Visual Studio Code displays the Workspace Trust prompt.

You created this empty folder yourself. Trusting it enables the editor features needed for the project.

• Open all eight starter files in Visual Studio Code by running this command from PowerShell:

code pyproject.toml .gitignore .env.example NOTES.md newsroom.py server.py test_newsroom.py README.md

What does this command do?

Visual Studio Code opens one tab for each starter file. Each file becomes visible in the Explorer after you save its content.

• Switch to the pyproject.toml tab.
• Replace its contents with this pinned project definition:

[project]
name = "newsroom-evidence-finder"
version = "0.1.0"
description = "Evidence-first public-data tooling for investigative reporters"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
    "anthropic==1.15.0",
    "duckdb==1.5.6",
    "mcp[cli]==2.3.0",
    "pydantic==2.14.1",
    "python-dotenv==1.2.4",
]

What does this configuration do?

• The project metadata names the repository plus its README.
• The Python requirement prevents uv from selecting an incompatible interpreter.
• The dependency list pins every package to the tested version used by this project.

• Press Ctrl+S to save pyproject.toml.

You should see pyproject.toml in the Explorer sidebar.

• Switch to the .gitignore tab.
• Replace its contents with these ignore rules:

.venv/
__pycache__/
*.pyc
*.duckdb
.env
.env.*
!.env.example

What do these rules protect?

Git ignores the local environment, generated Python files, DuckDB databases, and private environment files. The final rule keeps .env.example available as a safe public template.

• Press Ctrl+S to save .gitignore.

You should see .gitignore in the Explorer sidebar.

• Switch to the .env.example tab.
• Replace its contents with this safe credential template:

ANTHROPIC_API_KEY=your-api-key-here

Why keep a credential template?

The template documents the environment variable expected by the Anthropic SDK. Its placeholder contains no live credential.

• Press Ctrl+S to save .env.example.

You should see .env.example in the Explorer sidebar.

• Switch to the NOTES.md tab.
• Replace its contents with this reflection structure:

# Build Notes

For each step, add one line in your own words using this format:

`Step N | Did: ... | Broke: ... | Learned: ...`

## Entries

<!-- Type your Step 1 through Secret Mission entries here. Do not copy supplied reflections. -->

Why keep build notes?

The file gives you a consistent place for learner-written reflections. Those entries become evidence for later interview stories.

• Press Ctrl+S to save NOTES.md.

You should see NOTES.md in the Explorer sidebar.

• Switch to the newsroom.py tab.
• Replace its contents with this pipeline placeholder:

Initial placeholder file created for the newsroom evidence pipeline. It will define DB_PATH, CONTRACTS_FINDER_URL, COMPANIES_HOUSE_URL, COMPANIES_HOUSE_API_ROOT, CLAUDE_MODEL, EntityMention, EntityExtraction, connect, init_db, load_synthetic, count_exact_synthetic_links, normalise_name, count_normalised_synthetic_links, public-data ingestion functions, quote validation, extraction functions, search_records, trace_entity, and main in later steps.

What does this placeholder reserve?

The file reserves the project location for database access, public-data ingestion, entity extraction, search, and provenance tracing. Later steps replace the placeholder with working Python.

• Press Ctrl+S to save newsroom.py.

You should see newsroom.py in the Explorer sidebar.

• Switch to the server.py tab.
• Replace its contents with this MCP adapter placeholder:

Initial placeholder file created for the MCP adapter. It will expose search_records and trace_entity through MCPServer in Step 5.

What does this placeholder reserve?

The file reserves a separate adapter for MCP tools. Keeping that adapter separate allows the underlying newsroom functions to remain directly testable.

• Press Ctrl+S to save server.py.

You should see server.py in the Explorer sidebar.

• Switch to the test_newsroom.py tab.
• Replace its contents with this test-suite placeholder:

Initial placeholder file created for deterministic EvidenceFinderTests. It will test exact-name linking, normalized linking, source preservation, and quote validation in Step 5.

What does this placeholder reserve?

The file reserves deterministic checks for linking plus evidence preservation. Those tests will run without live APIs or API credits.

• Press Ctrl+S to save test_newsroom.py.

You should see test_newsroom.py in the Explorer sidebar.

• Switch to the README.md tab.
• Replace its contents with this project introduction:

# Newsroom Evidence Finder

A local Python pipeline that helps investigative reporters search UK public-contract records, reconcile company names, and trace AI-extracted entities back to evidence.

What does this introduction establish?

The README states the newsroom problem in one sentence. It also makes evidence traceability part of the project description from the first commit.

• Press Ctrl+S to save README.md.

You should see all eight starter files in the Explorer sidebar.

✔️ Awesome, I've got everything!

Your repository now has the exact starter files needed for the environment setup. Double check that every tab is saved.

ⓧ I'd like to double check the full code

Compare each saved file with the complete starter state below:

ANTHROPIC_API_KEY=your-api-key-here

.venv/
__pycache__/
*.pyc
*.duckdb
.env
.env.*
!.env.example

# Build Notes

For each step, add one line in your own words using this format:

`Step N | Did: ... | Broke: ... | Learned: ...`

## Entries

<!-- Type your Step 1 through Secret Mission entries here. Do not copy supplied reflections. -->

# Newsroom Evidence Finder

A local Python pipeline that helps investigative reporters search UK public-contract records, reconcile company names, and trace AI-extracted entities back to evidence.

Initial placeholder file created for the newsroom evidence pipeline. It will define DB_PATH, CONTRACTS_FINDER_URL, COMPANIES_HOUSE_URL, COMPANIES_HOUSE_API_ROOT, CLAUDE_MODEL, EntityMention, EntityExtraction, connect, init_db, load_synthetic, count_exact_synthetic_links, normalise_name, count_normalised_synthetic_links, public-data ingestion functions, quote validation, extraction functions, search_records, trace_entity, and main in later steps.

[project]
name = "newsroom-evidence-finder"
version = "0.1.0"
description = "Evidence-first public-data tooling for investigative reporters"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
    "anthropic==1.15.0",
    "duckdb==1.5.6",
    "mcp[cli]==2.3.0",
    "pydantic==2.14.1",
    "python-dotenv==1.2.4",
]

Initial placeholder file created for the MCP adapter. It will expose search_records and trace_entity through MCPServer in Step 5.

Initial placeholder file created for deterministic EvidenceFinderTests. It will test exact-name linking, normalized linking, source preservation, and quote validation in Step 5.

Learners who installed Python through uv need to select that interpreter for this repository. Learners who already had Python 3.11 or later can continue directly to synchronization.

✔️ Python already met the requirement

No extra interpreter pin is required. The Python requirement in pyproject.toml guides uv during synchronization.

ⓧ I installed Python through uv

• Pin Python 3.11 for this repository by running:

uv python pin 3.11

What does this command do?

uv records Python 3.11 as the interpreter choice for the current project. Future uv commands use that managed interpreter.

Dependency synchronization creates the isolated environment plus installs every exact version from pyproject.toml.

• Create .venv plus install the pinned packages by running:

uv sync

What does this command do?

uv reads the project metadata. It creates the local environment plus installs the pinned Anthropic SDK, DuckDB, MCP SDK, Pydantic, and python-dotenv packages.

The first synchronization can take a few minutes while uv downloads Python packages. PowerShell prints package activity during the process.

Once it finishes, you should see .venv in the Visual Studio Code Explorer. The terminal output should show the resolved and installed packages.

Did synchronization fail?

Confirm that PowerShell is inside the Desktop newsroom-evidence-finder folder. Also confirm that pyproject.toml matches the reference above.

If a package still fails to install, help me debug uv sync on Windows.

Create API keys and protect the Claude secret

The pipeline uses one key for Companies House enrichment plus one key for Claude extraction. The Companies House key is entered through getpass when needed. The Claude key stays in a local ignored file.

Credential screens deserve extra care because a copied key can authorize requests from anyone who sees it. Store each key privately before leaving its creation screen.

• Visit the Companies House Developer Hub.
• Register a Companies House user account.
• Create a live application.
• Select Create new key inside the application.
• Choose the client type requested by the Developer Hub.
• Create the API key.
• Store the Companies House API key in your password manager.

How will this key be used?

The ingestion pipeline requests the Companies House key through getpass. The key stays out of the source files plus the repository history.

• Open the official Claude API key guide.
• Sign in to Claude Console.
• Create a Claude API key from the API keys page.
• Copy the full key into your password manager before closing the one-time view.

The full Claude key is shown once. Keep it out of screenshots, chat messages, source files, and terminal history.

• Create a new file named .env inside the Visual Studio Code Explorer.
• Add this local environment entry:

ANTHROPIC_API_KEY=your-api-key-here

How does this entry work?

python-dotenv loads the local file into the process environment. The Anthropic SDK reads ANTHROPIC_API_KEY from that environment.

• Replace your-api-key-here with the Claude API key from your password manager.
• Press Ctrl+S to save .env.

• Verify that Git ignores the local credential file by running:

git check-ignore .env

What does this command prove?

Git prints paths that match an active ignore rule. Seeing .env confirms that the credential file is excluded from version control.

You should see .env as the command output. Your live Claude key now stays outside Git tracking.

Did Git print nothing?

Confirm that the file is named exactly .env. Confirm that the .env ignore rule is saved inside .gitignore.

If the file still appears trackable, help me verify my Git ignore rules.

Your final check loads every installed Python package without making a network request. Before you run it, which output would prove that the pinned environment is ready?

• Verify the complete Python toolchain by running:

uv run python -c "import anthropic, duckdb, mcp, pydantic; from dotenv import load_dotenv; print('toolchain ready')"

What does this verification cover?

uv runs Python inside the synchronized project environment. Python imports the five required packages before printing the success message.

The command does not call Companies House or Claude. It confirms local imports without using API credits.

You should see toolchain ready without an import error. That confirms the pinned environment can load every dependency.

Seeing an import error?

Run the synchronization again from the folder containing pyproject.toml. Confirm that the failing package appears in the pinned dependency list.

If the import still fails, help me diagnose the missing project dependency.

Your toolchain is reproducible plus your credentials are protected. Next, you will load synthetic newsroom records and watch an exact-name link fail.

Expose the Missed Synthetic Link

Your pinned environment is ready. You can now build the first runnable slice of the evidence finder.

Public contracting data can name the same supplier differently from a company register. This step makes that entity resolution gap visible with a controlled fixture before live records add more variation.

In this step, get ready to:

• Create the local database schema for contracts, companies, and extracted entities.
• Load a synthetic contract and company with different supplier names.
• Run a literal name join to expose the missed link.

Create the DuckDB schema

DuckDB stores the synthetic evidence in a local file named newsroom.duckdb. The schema keeps contracts, company records, and future extractions in separate tables.

• Replace the placeholder content in newsroom.py with the imports and local configuration below:

from __future__ import annotations

import argparse
import base64
import getpass
import json
import re
from typing import Literal
from urllib.parse import urlencode
from urllib.request import Request, urlopen

import duckdb
from anthropic import Anthropic
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

DB_PATH = "newsroom.duckdb"

What Does This Foundation Do?

• The imports prepare newsroom.py for database work, command handling, public-data requests, and later extraction.
• load_dotenv() loads the ignored local environment file for later Claude access.
• DB_PATH gives every database function one consistent local destination.

• Save newsroom.py.
• Confirm your editor shows no Python syntax problems in the new foundation.

Seeing Import Problems?

Confirm VS Code is using the Python interpreter from the existing .venv folder. The pinned packages already exist in that environment.

Ask for help with the environment selection.

• Add the extraction data models below DB_PATH in newsroom.py by copying this code:

class EntityMention(BaseModel):
    entity_text: str
    entity_type: Literal["PERSON", "COMPANY"]
    source_quote: str


class EntityExtraction(BaseModel):
    entities: list[EntityMention]

Why Define These Models Now?

Pydantic models define the exact structure that a later extraction must follow. Keeping that contract beside the database setup gives the pipeline one shared definition of an entity mention.

• EntityMention requires the entity text, its allowed type, and a source quote.
• EntityExtraction wraps the resulting entity list.

• Save newsroom.py.
• Confirm the editor recognizes both class definitions without a syntax problem.

Models Showing Syntax Problems?

Check that both classes begin at the left edge of the file. Confirm each field inside a class uses four spaces of indentation.

Ask for help checking the model definitions.

• Add the database connection and first table below EntityExtraction by copying this code:

def connect(db_path: str = DB_PATH) -> duckdb.DuckDBPyConnection:
    return duckdb.connect(db_path)


def init_db(con: duckdb.DuckDBPyConnection) -> None:
    con.execute(
        """
        CREATE TABLE IF NOT EXISTS contracts (
            source_id VARCHAR,
            title VARCHAR,
            description VARCHAR,
            buyer VARCHAR,
            supplier VARCHAR,
            value_gbp DOUBLE,
            source_url VARCHAR,
            source_name VARCHAR
        )
        """
    )

How Does the Contract Table Help?

• connect() opens the database at the path held by DB_PATH.
• init_db() accepts that connection so every table is created in the same database.
• contracts keeps the reporting fields beside the source identity and source reference.

• Save newsroom.py.
• Confirm the contracts definition contains eight named columns.

Contract Schema Looks Misaligned?

Keep the SQL text inside the triple quotes. Keep the closing parenthesis for con.execute() aligned with its opening call.

Ask for help checking the schema structure.

• Add the company table directly after the first con.execute() call inside init_db() by copying this code:

    con.execute(
        """
        CREATE TABLE IF NOT EXISTS companies (
            company_number VARCHAR,
            company_name VARCHAR,
            company_status VARCHAR,
            address VARCHAR,
            source_url VARCHAR
        )
        """
    )

Why Keep Companies Separate?

The companies table represents register data independently from procurement records. This separation lets the pipeline compare supplier names while preserving each source.

• Save newsroom.py.
• Confirm the new con.execute() call remains indented inside init_db().

Company Table Outside the Function?

The new block needs four leading spaces before its first con.execute() line. That indentation keeps it inside init_db().

Ask for help fixing the table placement.

• Add the extraction table directly after the company table inside init_db() by copying this code:

    con.execute(
        """
        CREATE TABLE IF NOT EXISTS extracted_entities (
            source_id VARCHAR,
            entity_text VARCHAR,
            entity_type VARCHAR,
            source_quote VARCHAR,
            source_text VARCHAR,
            matched_company_number VARCHAR,
            match_status VARCHAR,
            model_id VARCHAR
        )
        """
    )

How Does This Preserve Evidence?

The extracted_entities table reserves fields for the exact source quote and complete source text. It also keeps the originating record ID and extraction model ID.

Those fields establish provenance for every future extraction. A reporter can trace a result back to the stored evidence.

• Save newsroom.py.
• Confirm init_db() now contains table definitions for contracts, companies, and extracted_entities.

Missing a Closing Parenthesis?

Compare the closing triple quotes and parentheses across all three table blocks. Each con.execute() call needs its own closing parenthesis.

Ask for help checking the three table calls.

Load the synthetic evidence

A synthetic fixture gives you a stable example with known facts. The contract names Northstar Analytics while the company record uses NORTHSTAR ANALYTICS LTD.

• Add load_synthetic() below init_db() in newsroom.py by copying this code:

def load_synthetic(con: duckdb.DuckDBPyConnection) -> None:
    con.execute("DELETE FROM contracts WHERE source_id = ?", ["synthetic-contract-001"])
    con.execute("DELETE FROM companies WHERE company_number = ?", ["SYN-001"])
    con.execute(
        "INSERT INTO contracts VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
        [
            "synthetic-contract-001",
            "Council analytics review",
            "Reporter note: Maya Chen says Northstar Analytics won the analytics work.",
            "Riverton Council",
            "Northstar Analytics",
            125000.0,
            "local fixture",
            "Synthetic fixture",
        ],
    )
    con.execute(
        "INSERT INTO companies VALUES (?, ?, ?, ?, ?)",
        [
            "SYN-001",
            "NORTHSTAR ANALYTICS LTD",
            "active",
            "1 Civic Way, Riverton",
            "local fixture",
        ],
    )

What Does the Fixture Preserve?

• The initial deletion queries make repeated runs produce one copy of each synthetic record.
• The contract stores a title, reporter note, buyer, supplier, value, and local source reference.
• The company stores its register-style name beside a synthetic company number and local source reference.
• The two supplier names describe the same fictional business with different strings.

• Save newsroom.py.
• Confirm the fixture retains local fixture in both source records.

Fixture Showing a Syntax Problem?

Check the commas between every list value. Confirm the contract has eight values while the company has five.

Ask for help checking the insert values.

• Add the literal join and synthetic command below load_synthetic() by copying this code:

def count_exact_synthetic_links(con: duckdb.DuckDBPyConnection) -> int:
    row = con.execute(
        """
        SELECT COUNT(*)
        FROM contracts AS c
        JOIN companies AS co ON c.supplier = co.company_name
        WHERE c.source_id = ?
        """,
        ["synthetic-contract-001"],
    ).fetchone()
    return int(row[0]) if row else 0


def main() -> None:
    parser = argparse.ArgumentParser(description="Newsroom Evidence Finder")
    subparsers = parser.add_subparsers(dest="command", required=True)
    subparsers.add_parser("synthetic")
    args = parser.parse_args()

    with connect() as con:
        init_db(con)
        if args.command == "synthetic":
            load_synthetic(con)
            print(f"Exact-name links: {count_exact_synthetic_links(con)}")


if __name__ == "__main__":
    main()

How Does the Exact Join Work?

• count_exact_synthetic_links() compares supplier directly with company_name.
• The WHERE condition limits the count to the controlled synthetic contract.
• main() creates the schema before loading the fixture.
• The synthetic command prints the resulting exact-match count.

• Save newsroom.py.
• Confirm the final two lines remain at the left edge except for the indented call to main().

Command Logic Showing a Syntax Problem?

Confirm the SQL remains inside triple quotes. Check that the final main() call has four leading spaces.

Ask for help checking the command structure.

✔️ Awesome, I've got everything!

Great. Double check that you saved newsroom.py before running the synthetic pipeline.

ⓧ I'd like to double check the full code

Compare your complete newsroom.py file with this reference:

from __future__ import annotations

import argparse
import base64
import getpass
import json
import re
from typing import Literal
from urllib.parse import urlencode
from urllib.request import Request, urlopen

import duckdb
from anthropic import Anthropic
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

DB_PATH = "newsroom.duckdb"


class EntityMention(BaseModel):
    entity_text: str
    entity_type: Literal["PERSON", "COMPANY"]
    source_quote: str


class EntityExtraction(BaseModel):
    entities: list[EntityMention]


def connect(db_path: str = DB_PATH) -> duckdb.DuckDBPyConnection:
    return duckdb.connect(db_path)


def init_db(con: duckdb.DuckDBPyConnection) -> None:
    con.execute(
        """
        CREATE TABLE IF NOT EXISTS contracts (
            source_id VARCHAR,
            title VARCHAR,
            description VARCHAR,
            buyer VARCHAR,
            supplier VARCHAR,
            value_gbp DOUBLE,
            source_url VARCHAR,
            source_name VARCHAR
        )
        """
    )
    con.execute(
        """
        CREATE TABLE IF NOT EXISTS companies (
            company_number VARCHAR,
            company_name VARCHAR,
            company_status VARCHAR,
            address VARCHAR,
            source_url VARCHAR
        )
        """
    )
    con.execute(
        """
        CREATE TABLE IF NOT EXISTS extracted_entities (
            source_id VARCHAR,
            entity_text VARCHAR,
            entity_type VARCHAR,
            source_quote VARCHAR,
            source_text VARCHAR,
            matched_company_number VARCHAR,
            match_status VARCHAR,
            model_id VARCHAR
        )
        """
    )


def load_synthetic(con: duckdb.DuckDBPyConnection) -> None:
    con.execute("DELETE FROM contracts WHERE source_id = ?", ["synthetic-contract-001"])
    con.execute("DELETE FROM companies WHERE company_number = ?", ["SYN-001"])
    con.execute(
        "INSERT INTO contracts VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
        [
            "synthetic-contract-001",
            "Council analytics review",
            "Reporter note: Maya Chen says Northstar Analytics won the analytics work.",
            "Riverton Council",
            "Northstar Analytics",
            125000.0,
            "local fixture",
            "Synthetic fixture",
        ],
    )
    con.execute(
        "INSERT INTO companies VALUES (?, ?, ?, ?, ?)",
        [
            "SYN-001",
            "NORTHSTAR ANALYTICS LTD",
            "active",
            "1 Civic Way, Riverton",
            "local fixture",
        ],
    )


def count_exact_synthetic_links(con: duckdb.DuckDBPyConnection) -> int:
    row = con.execute(
        """
        SELECT COUNT(*)
        FROM contracts AS c
        JOIN companies AS co ON c.supplier = co.company_name
        WHERE c.source_id = ?
        """,
        ["synthetic-contract-001"],
    ).fetchone()
    return int(row[0]) if row else 0


def main() -> None:
    parser = argparse.ArgumentParser(description="Newsroom Evidence Finder")
    subparsers = parser.add_subparsers(dest="command", required=True)
    subparsers.add_parser("synthetic")
    args = parser.parse_args()

    with connect() as con:
        init_db(con)
        if args.command == "synthetic":
            load_synthetic(con)
            print(f"Exact-name links: {count_exact_synthetic_links(con)}")


if __name__ == "__main__":
    main()

Run the exact-name join

The fixture now gives the join two names that refer to the same fictional supplier. The command tests whether literal string equality can recognize that relationship.

Before you run it, predict whether the two stored names are equal character for character.

• Run the synthetic pipeline in the PowerShell terminal from the existing newsroom-evidence-finder folder by using this command:

uv run python newsroom.py synthetic

You will see Exact-name links: 0 in the terminal. The exact join misses the relationship because the two stored names are different strings.

You have made a real data-quality failure visible. The zero proves that a successful query can still produce an incomplete answer.

Why Is Zero the Intended Result?

The contract stores Northstar Analytics. The company record stores NORTHSTAR ANALYTICS LTD.

Literal equality compares every character. Capitalization and the company suffix prevent a match even though a reporter would recognize the shared identity.

Did the Synthetic Command Fail?

Confirm the terminal is still inside the existing newsroom-evidence-finder folder. Check that newsroom.py is saved before running the command again.

Ask for help debugging the synthetic run.

• Capture the terminal showing Exact-name links: 0 with your usual Windows screenshot tool.
• Save the missed-link screenshot for your project evidence.
• Add a blank line below the existing description in README.md.
• Add this sentence as a new paragraph: The synthetic fixture intentionally demonstrates that an exact supplier-name join misses Northstar Analytics versus NORTHSTAR ANALYTICS LTD.
• Replace the comment under Entries in NOTES.md with your personal Step 1 and Step 2 entries.
• Use the existing Step N | Did: ... | Broke: ... | Learned: ... format for your Step 2 reflection.
• Save README.md.
• Save NOTES.md.

Your pipeline now exposes the name mismatch instead of hiding it. Next up, you will resolve that alias and apply the same evidence-first approach to current public records.

Resolve Aliases and Ingest Public Data

You captured the exact-name failure in the previous step. That result proved your DuckDB tables can preserve evidence while still missing an obvious company relationship.

This step uses entity normalization to resolve the synthetic alias. You will then test the same evidence-first approach against current public records.

In this step, get ready to:

• Normalize company names so the synthetic alias resolves.
• Load current award records from Contracts Finder.
• Enrich one supplier through the Companies House Public Data API.

Normalize supplier and company names

Real organizations appear under abbreviations or legal names. A stable comparison key removes formatting differences before the pipeline attempts a match.

• In newsroom.py, find load_synthetic().
• Add normalise_name() directly above load_synthetic() by typing this function:

def normalise_name(value: str) -> str:
    cleaned = re.sub(r"[^A-Z0-9]+", " ", value.upper()).strip()
    tokens = cleaned.split()
    suffixes = {"LTD", "LIMITED", "LLP", "PLC"}
    while tokens and tokens[-1] in suffixes:
        tokens.pop()
    return " ".join(tokens)

What does this function do?

• The regular expression replaces punctuation with spaces after converting the name to uppercase.
• The token list lets the function remove trailing legal suffixes without changing words inside the company name.
• The final join collapses repeated whitespace into one consistent comparison key.

• Save newsroom.py.
• Check the normalized Northstar name by running this command in the PowerShell panel from earlier:

uv run python -c "from newsroom import normalise_name; print(normalise_name('NORTHSTAR ANALYTICS LTD'))"

What does this check prove?

The command imports your new function through the managed environment. It prints the comparison key produced from the registered company name.

You should see NORTHSTAR ANALYTICS. The legal suffix has disappeared from the value used for matching.

Name still includes the suffix?

Check that suffixes contains LTD in uppercase. Confirm the while loop appears before the return statement.

Ask for help if the function produces a different key: help me debug my company-name normalization function

• In newsroom.py, find count_exact_synthetic_links().
• Add the normalized comparison directly below that function by typing:

def count_normalised_synthetic_links(con: duckdb.DuckDBPyConnection) -> int:
    supplier_row = con.execute(
        "SELECT supplier FROM contracts WHERE source_id = ?",
        ["synthetic-contract-001"],
    ).fetchone()
    company_rows = con.execute(
        "SELECT company_name FROM companies WHERE company_number = ?",
        ["SYN-001"],
    ).fetchall()
    if not supplier_row:
        return 0
    supplier_key = normalise_name(supplier_row[0])
    return sum(normalise_name(row[0]) == supplier_key for row in company_rows)

How does the new comparison work?

• The parameterized queries retrieve the synthetic supplier plus its candidate company record.
• The contract name becomes a normalized key before comparison.
• Each company name passes through the same function so both sides follow one rule.

• Save newsroom.py.
• Confirm the new function has no syntax errors by rerunning the normalization check from above.

The synthetic contract now has a comparison path that treats Northstar Analytics as the same key as NORTHSTAR ANALYTICS LTD.

Fetch and preserve Contracts Finder awards

The live batch introduces optional fields that a synthetic fixture cannot reproduce. Defensive parsing keeps usable awards when a description or document link is absent.

• In newsroom.py, find the DB_PATH declaration.
• Add the public-data endpoint constants directly below it by typing:

CONTRACTS_FINDER_URL = (
    "https://www.contractsfinder.service.gov.uk/Published/Notices/OCDS/Search"
)
COMPANIES_HOUSE_URL = (
    "https://api.company-information.service.gov.uk/search/companies"
)
COMPANIES_HOUSE_API_ROOT = "https://api.company-information.service.gov.uk"

Why keep the endpoints together?

Named constants give each request one canonical endpoint. They also keep source-reference construction consistent throughout the pipeline.

• Return to the area below count_normalised_synthetic_links().
• Add the shared JSON request helper by typing:

def fetch_json(url: str, headers: dict[str, str] | None = None) -> dict:
    request = Request(url, headers=headers or {"Accept": "application/json"})
    with urlopen(request, timeout=30) as response:
        return json.load(response)

What does the request helper handle?

The helper creates one request path for both public-data services. Its timeout prevents a stalled network request from holding the pipeline indefinitely.

• Add the document-link helper directly below fetch_json() by typing:

def first_document_url(documents: list[dict]) -> str:
    for document in documents:
        url = document.get("url", "")
        if url:
            return str(url)
    return ""

Why isolate document selection?

An award can contain several documents or none. This helper returns the first usable URL while giving the caller a safe empty-string fallback.

• Start fetch_contracts_finder_awards() directly below first_document_url() by typing this first section:

def fetch_contracts_finder_awards(limit: int = 5) -> list[dict[str, object]]:
    query = urlencode({"stages": "award", "limit": str(limit)})
    payload = fetch_json(f"{CONTRACTS_FINDER_URL}?{query}")
    records: list[dict[str, object]] = []

    for release in payload.get("releases", []):
        tender = release.get("tender", {}) or {}
        awards = release.get("awards", []) or []
        award = awards[0] if awards else {}
        suppliers = award.get("suppliers", []) or []
        supplier = str(suppliers[0].get("name", "")).strip() if suppliers else ""
        if not supplier:
            continue

        value = award.get("value", {}) or tender.get("value", {}) or {}
        source_url = first_document_url(award.get("documents", []) or [])
        if not source_url:
            source_url = first_document_url(tender.get("documents", []) or [])
        if not source_url:
            source_url = str(payload.get("uri", CONTRACTS_FINDER_URL))

How does the parser tolerate missing fields?

Each optional object falls back to an empty dictionary or list. Awards without suppliers are skipped because they cannot support company enrichment.

The source reference prefers an award document. It falls back to a tender document or the API package URL.

• Continue inside fetch_contracts_finder_awards() directly after the final source URL assignment by typing:

        records.append(
            {
                "source_id": str(release.get("id") or release.get("ocid") or ""),
                "title": str(tender.get("title", "")),
                "description": str(tender.get("description", "")),
                "buyer": str((release.get("buyer", {}) or {}).get("name", "")),
                "supplier": supplier,
                "value_gbp": float(value.get("amount", 0) or 0),
                "source_url": source_url,
                "source_name": "Contracts Finder",
            }
        )

    if not records:
        raise RuntimeError("Contracts Finder returned no award records with suppliers.")
    return records

What does each flattened record preserve?

The dictionary turns nested procurement data into columns that reporters can search later. The source ID plus source URL retain the path back to the public evidence.

• Add save_contracts() directly below the completed fetch function by typing:

def save_contracts(con: duckdb.DuckDBPyConnection, records: list[dict[str, object]]) -> None:
    for record in records:
        con.execute("DELETE FROM contracts WHERE source_id = ?", [record["source_id"]])
        con.execute(
            "INSERT INTO contracts VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            [
                record["source_id"],
                record["title"],
                record["description"],
                record["buyer"],
                record["supplier"],
                record["value_gbp"],
                record["source_url"],
                record["source_name"],
            ],
        )

Why delete before inserting?

The source ID identifies the same public release across repeated runs. Removing an existing row first prevents duplicate copies while allowing refreshed values to replace it.

• Save newsroom.py.
• Fetch the current award batch by running this command:

uv run python -c "from newsroom import connect, fetch_contracts_finder_awards, save_contracts; con = connect(); records = fetch_contracts_finder_awards(); save_contracts(con, records); print(len(records))"

What does this live check do?

The command fetches up to five current award releases. It writes the flattened records to the existing database before printing the number saved.

You should see a number from 1 to 5. Your database now contains live contracts with public source references.

Live award fetch failed?

Confirm your internet connection can reach Contracts Finder. Run the command again if the service returned a temporary network response.

Ask for help with the request if it continues to fail: help me diagnose my Contracts Finder award request

Enrich a supplier and verify the resolved link

A procurement supplier name becomes more useful when it connects to a registered company. The enrichment stage searches suppliers in order while retaining every contract that was already saved.

• Add the first section of fetch_companies_house_matches() directly below save_contracts() by typing:

def fetch_companies_house_matches(
    query: str, api_key: str, items_per_page: int = 5
) -> list[dict[str, str]]:
    credentials = base64.b64encode(f"{api_key}:".encode()).decode()
    params = urlencode({"q": query, "items_per_page": str(items_per_page)})
    payload = fetch_json(
        f"{COMPANIES_HOUSE_URL}?{params}",
        {
            "Accept": "application/json",
            "Authorization": f"Basic {credentials}",
        },
    )
    matches: list[dict[str, str]] = []

How is the request authenticated?

HTTP Basic authentication uses the API key as the username. The colon after the key represents the required blank password before Base64 encoding.

• Continue inside fetch_companies_house_matches() directly after the empty matches list by typing:

    for item in payload.get("items", []):
        company_number = str(item.get("company_number", ""))
        relative_url = str((item.get("links", {}) or {}).get("self", ""))
        matches.append(
            {
                "company_number": company_number,
                "company_name": str(item.get("title", "")),
                "company_status": str(item.get("company_status", "")),
                "address": str(item.get("address_snippet", "")),
                "source_url": (
                    f"{COMPANIES_HOUSE_API_ROOT}{relative_url}"
                    if relative_url
                    else f"{COMPANIES_HOUSE_URL}?{params}"
                ),
            }
        )
    return matches

What comes back from company search?

Each result retains its registered name plus company number. The status and address help a reporter distinguish similar candidates.

The source URL points to the company resource when the response provides one. The search request remains as a fallback reference.

• Add save_companies() directly below the search function by typing:

def save_companies(con: duckdb.DuckDBPyConnection, matches: list[dict[str, str]]) -> None:
    for match in matches:
        con.execute(
            "DELETE FROM companies WHERE company_number = ?",
            [match["company_number"]],
        )
        con.execute(
            "INSERT INTO companies VALUES (?, ?, ?, ?, ?)",
            [
                match["company_number"],
                match["company_name"],
                match["company_status"],
                match["address"],
                match["source_url"],
            ],
        )

How are company matches stored?

The company number acts as the stable identifier for replacement. Every saved row retains the public source URL used for later verification.

• Add ingest_live_data() directly below save_companies() by typing:

def ingest_live_data(con: duckdb.DuckDBPyConnection) -> tuple[int, int, str]:
    contracts = fetch_contracts_finder_awards()
    save_contracts(con, contracts)

    api_key = getpass.getpass("Companies House API key: ").strip()
    if not api_key:
        raise RuntimeError("A Companies House API key is required for enrichment.")

    company_count = 0
    matched_supplier = ""
    for contract in contracts:
        matches = fetch_companies_house_matches(str(contract["supplier"]), api_key)
        if matches:
            save_companies(con, matches)
            company_count = len(matches)
            matched_supplier = str(contract["supplier"])
            break

    if not matched_supplier:
        raise RuntimeError("No Companies House match was found for the fetched suppliers.")
    return len(contracts), company_count, matched_supplier

Why try suppliers in sequence?

A public contract remains valuable even when its first supplier search returns no company. The loop keeps every contract while trying the next supplier for an enrichment candidate.

The key exists only in memory during the request. It is never written into newsroom.py or the database.

• In main(), locate these parser lines:

    subparsers = parser.add_subparsers(dest="command", required=True)
    subparsers.add_parser("synthetic")

What are you locating?

These lines currently register the only command your script accepts. The next edit adds a separate path for live ingestion.

• Replace those parser lines with this version:

    subparsers = parser.add_subparsers(dest="command", required=True)
    subparsers.add_parser("synthetic")
    subparsers.add_parser("ingest")

What changed in the parser?

The script now recognizes ingest as a valid command while preserving the original synthetic demonstration.

• Near the bottom of main(), locate the existing synthetic branch:

        if args.command == "synthetic":
            load_synthetic(con)
            print(f"Exact-name links: {count_exact_synthetic_links(con)}")

What does the current branch prove?

The original branch preserves the designed obstacle. Running the synthetic command still demonstrates that literal equality misses the alias.

• Replace that branch with the completed command handling:

        if args.command == "synthetic":
            load_synthetic(con)
            print(f"Exact-name links: {count_exact_synthetic_links(con)}")
        elif args.command == "ingest":
            load_synthetic(con)
            print(
                f"Normalised synthetic links: {count_normalised_synthetic_links(con)}"
            )
            contract_count, company_count, supplier = ingest_live_data(con)
            print(f"Contracts Finder records: {contract_count}")
            print(f"Companies House matches: {company_count}")
            print(f"Enriched supplier: {supplier}")

What does the ingest command report?

The command first proves the synthetic alias resolves. It then reports the number of live contracts plus the first supplier with saved company matches.

• Save newsroom.py.

✔️ Awesome, I've got everything!

Great. Double-check that newsroom.py is saved before you run the live ingestion.

ⓧ I'd like to double check the full code

Compare your cumulative newsroom.py file with this reference:

from __future__ import annotations

import argparse
import base64
import getpass
import json
import re
from typing import Literal
from urllib.parse import urlencode
from urllib.request import Request, urlopen

import duckdb
from anthropic import Anthropic
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

DB_PATH = "newsroom.duckdb"
CONTRACTS_FINDER_URL = (
    "https://www.contractsfinder.service.gov.uk/Published/Notices/OCDS/Search"
)
COMPANIES_HOUSE_URL = (
    "https://api.company-information.service.gov.uk/search/companies"
)
COMPANIES_HOUSE_API_ROOT = "https://api.company-information.service.gov.uk"


class EntityMention(BaseModel):
    entity_text: str
    entity_type: Literal["PERSON", "COMPANY"]
    source_quote: str


class EntityExtraction(BaseModel):
    entities: list[EntityMention]


def connect(db_path: str = DB_PATH) -> duckdb.DuckDBPyConnection:
    return duckdb.connect(db_path)


def init_db(con: duckdb.DuckDBPyConnection) -> None:
    con.execute(
        """
        CREATE TABLE IF NOT EXISTS contracts (
            source_id VARCHAR,
            title VARCHAR,
            description VARCHAR,
            buyer VARCHAR,
            supplier VARCHAR,
            value_gbp DOUBLE,
            source_url VARCHAR,
            source_name VARCHAR
        )
        """
    )
    con.execute(
        """
        CREATE TABLE IF NOT EXISTS companies (
            company_number VARCHAR,
            company_name VARCHAR,
            company_status VARCHAR,
            address VARCHAR,
            source_url VARCHAR
        )
        """
    )
    con.execute(
        """
        CREATE TABLE IF NOT EXISTS extracted_entities (
            source_id VARCHAR,
            entity_text VARCHAR,
            entity_type VARCHAR,
            source_quote VARCHAR,
            source_text VARCHAR,
            matched_company_number VARCHAR,
            match_status VARCHAR,
            model_id VARCHAR
        )
        """
    )


def normalise_name(value: str) -> str:
    cleaned = re.sub(r"[^A-Z0-9]+", " ", value.upper()).strip()
    tokens = cleaned.split()
    suffixes = {"LTD", "LIMITED", "LLP", "PLC"}
    while tokens and tokens[-1] in suffixes:
        tokens.pop()
    return " ".join(tokens)


def load_synthetic(con: duckdb.DuckDBPyConnection) -> None:
    con.execute("DELETE FROM contracts WHERE source_id = ?", ["synthetic-contract-001"])
    con.execute("DELETE FROM companies WHERE company_number = ?", ["SYN-001"])
    con.execute(
        "INSERT INTO contracts VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
        [
            "synthetic-contract-001",
            "Council analytics review",
            "Reporter note: Maya Chen says Northstar Analytics won the analytics work.",
            "Riverton Council",
            "Northstar Analytics",
            125000.0,
            "local fixture",
            "Synthetic fixture",
        ],
    )
    con.execute(
        "INSERT INTO companies VALUES (?, ?, ?, ?, ?)",
        [
            "SYN-001",
            "NORTHSTAR ANALYTICS LTD",
            "active",
            "1 Civic Way, Riverton",
            "local fixture",
        ],
    )


def count_exact_synthetic_links(con: duckdb.DuckDBPyConnection) -> int:
    row = con.execute(
        """
        SELECT COUNT(*)
        FROM contracts AS c
        JOIN companies AS co ON c.supplier = co.company_name
        WHERE c.source_id = ?
        """,
        ["synthetic-contract-001"],
    ).fetchone()
    return int(row[0]) if row else 0


def count_normalised_synthetic_links(con: duckdb.DuckDBPyConnection) -> int:
    supplier_row = con.execute(
        "SELECT supplier FROM contracts WHERE source_id = ?",
        ["synthetic-contract-001"],
    ).fetchone()
    company_rows = con.execute(
        "SELECT company_name FROM companies WHERE company_number = ?",
        ["SYN-001"],
    ).fetchall()
    if not supplier_row:
        return 0
    supplier_key = normalise_name(supplier_row[0])
    return sum(normalise_name(row[0]) == supplier_key for row in company_rows)


def fetch_json(url: str, headers: dict[str, str] | None = None) -> dict:
    request = Request(url, headers=headers or {"Accept": "application/json"})
    with urlopen(request, timeout=30) as response:
        return json.load(response)


def first_document_url(documents: list[dict]) -> str:
    for document in documents:
        url = document.get("url", "")
        if url:
            return str(url)
    return ""


def fetch_contracts_finder_awards(limit: int = 5) -> list[dict[str, object]]:
    query = urlencode({"stages": "award", "limit": str(limit)})
    payload = fetch_json(f"{CONTRACTS_FINDER_URL}?{query}")
    records: list[dict[str, object]] = []

    for release in payload.get("releases", []):
        tender = release.get("tender", {}) or {}
        awards = release.get("awards", []) or []
        award = awards[0] if awards else {}
        suppliers = award.get("suppliers", []) or []
        supplier = str(suppliers[0].get("name", "")).strip() if suppliers else ""
        if not supplier:
            continue

        value = award.get("value", {}) or tender.get("value", {}) or {}
        source_url = first_document_url(award.get("documents", []) or [])
        if not source_url:
            source_url = first_document_url(tender.get("documents", []) or [])
        if not source_url:
            source_url = str(payload.get("uri", CONTRACTS_FINDER_URL))

        records.append(
            {
                "source_id": str(release.get("id") or release.get("ocid") or ""),
                "title": str(tender.get("title", "")),
                "description": str(tender.get("description", "")),
                "buyer": str((release.get("buyer", {}) or {}).get("name", "")),
                "supplier": supplier,
                "value_gbp": float(value.get("amount", 0) or 0),
                "source_url": source_url,
                "source_name": "Contracts Finder",
            }
        )

    if not records:
        raise RuntimeError("Contracts Finder returned no award records with suppliers.")
    return records


def save_contracts(con: duckdb.DuckDBPyConnection, records: list[dict[str, object]]) -> None:
    for record in records:
        con.execute("DELETE FROM contracts WHERE source_id = ?", [record["source_id"]])
        con.execute(
            "INSERT INTO contracts VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            [
                record["source_id"],
                record["title"],
                record["description"],
                record["buyer"],
                record["supplier"],
                record["value_gbp"],
                record["source_url"],
                record["source_name"],
            ],
        )


def fetch_companies_house_matches(
    query: str, api_key: str, items_per_page: int = 5
) -> list[dict[str, str]]:
    credentials = base64.b64encode(f"{api_key}:".encode()).decode()
    params = urlencode({"q": query, "items_per_page": str(items_per_page)})
    payload = fetch_json(
        f"{COMPANIES_HOUSE_URL}?{params}",
        {
            "Accept": "application/json",
            "Authorization": f"Basic {credentials}",
        },
    )
    matches: list[dict[str, str]] = []
    for item in payload.get("items", []):
        company_number = str(item.get("company_number", ""))
        relative_url = str((item.get("links", {}) or {}).get("self", ""))
        matches.append(
            {
                "company_number": company_number,
                "company_name": str(item.get("title", "")),
                "company_status": str(item.get("company_status", "")),
                "address": str(item.get("address_snippet", "")),
                "source_url": (
                    f"{COMPANIES_HOUSE_API_ROOT}{relative_url}"
                    if relative_url
                    else f"{COMPANIES_HOUSE_URL}?{params}"
                ),
            }
        )
    return matches


def save_companies(con: duckdb.DuckDBPyConnection, matches: list[dict[str, str]]) -> None:
    for match in matches:
        con.execute(
            "DELETE FROM companies WHERE company_number = ?",
            [match["company_number"]],
        )
        con.execute(
            "INSERT INTO companies VALUES (?, ?, ?, ?, ?)",
            [
                match["company_number"],
                match["company_name"],
                match["company_status"],
                match["address"],
                match["source_url"],
            ],
        )


def ingest_live_data(con: duckdb.DuckDBPyConnection) -> tuple[int, int, str]:
    contracts = fetch_contracts_finder_awards()
    save_contracts(con, contracts)

    api_key = getpass.getpass("Companies House API key: ").strip()
    if not api_key:
        raise RuntimeError("A Companies House API key is required for enrichment.")

    company_count = 0
    matched_supplier = ""
    for contract in contracts:
        matches = fetch_companies_house_matches(str(contract["supplier"]), api_key)
        if matches:
            save_companies(con, matches)
            company_count = len(matches)
            matched_supplier = str(contract["supplier"])
            break

    if not matched_supplier:
        raise RuntimeError("No Companies House match was found for the fetched suppliers.")
    return len(contracts), company_count, matched_supplier


def main() -> None:
    parser = argparse.ArgumentParser(description="Newsroom Evidence Finder")
    subparsers = parser.add_subparsers(dest="command", required=True)
    subparsers.add_parser("synthetic")
    subparsers.add_parser("ingest")
    args = parser.parse_args()

    with connect() as con:
        init_db(con)
        if args.command == "synthetic":
            load_synthetic(con)
            print(f"Exact-name links: {count_exact_synthetic_links(con)}")
        elif args.command == "ingest":
            load_synthetic(con)
            print(
                f"Normalised synthetic links: {count_normalised_synthetic_links(con)}"
            )
            contract_count, company_count, supplier = ingest_live_data(con)
            print(f"Contracts Finder records: {contract_count}")
            print(f"Companies House matches: {company_count}")
            print(f"Enriched supplier: {supplier}")


if __name__ == "__main__":
    main()

This credential step is easy to second-guess. The getpass prompt hides the key while you type or paste it.

• Copy your live Companies House API key from the Developer Hub.

Before you run the pipeline, do you expect the normalized synthetic comparison to remain at zero or resolve to one match?

• Run the completed ingestion command in PowerShell:

uv run python newsroom.py ingest

What does the completed run prove?

The command reloads the synthetic fixture before applying the normalized comparison. It then fetches current award records plus company candidates.

The terminal waits at Companies House API key: for your credential. Your pasted characters remain hidden.

You should first see Normalised synthetic links: 1. You should then see at least one Contracts Finder record plus at least one Companies House match.

The final line names the supplier that produced the saved company candidates. The saved contract and company rows retain non-empty public source references.

No company match found?

Live procurement records change between runs. Retry the command to fetch a new small batch if none of the current supplier names produces a result.

Confirm that you pasted a live API key when the hidden prompt appeared. Do not add the key to newsroom.py.

Ask for targeted help if enrichment still fails: help me debug my Companies House enrichment request

• Save a screenshot that shows Normalised synthetic links: 1 plus the nonzero live record counts.
• Explain aloud how suffix removal improves matching while still allowing false positives.
• Name one optional live-data field that required a defensive fallback.
• Replace the final paragraph in README.md with: The pipeline uses a deliberately failed exact-name join before normalizing UK company suffixes. Contracts Finder award releases and Companies House search results are retained with source references.
• Add a personal Step 3 entry to NOTES.md using the existing reflection format.

That missed alias now resolves against a repeatable comparison key. Next, you will extract entities while preserving the exact evidence behind every result.

Extract Traceable Entities with Claude

Your DuckDB pipeline now links normalized supplier names to public company records. Important people or organizations can still remain buried inside contract text.

Claude can extract those names. An LLM result becomes useful to a reporter only when the original evidence remains attached.

You will control the cost before making a live request. You will also require every extracted entity to carry a literal source quote.

In this step, get ready to:

• Check the available Claude credits before setting a small spend limit.
• Build a structured extraction contract with exact-quote validation.
• Run the optional extraction before recording the result in your project notes.

Control the live API cost

A live Claude request uses API credits. Checking the balance first keeps this experiment within a limit you choose.

• Return to the Claude Console from your earlier setup.
• Select Settings in the console navigation.
• Select Billing to answer whether your organization has credits available.
• Read the value under Credit balance.

Why check the actual balance?

API access depends on the credits available to your organization. The balance shown in the console is the source of truth for this live call.

• Select Set limit under Spend limits if no custom organization limit exists.
• Select Adjust limit if an organization limit already exists.
• Enter a small organization limit for this one-call experiment.
• Confirm the chosen limit appears under Spend limits.

The displayed limit now caps the organization spend. Next, check whether automatic credit purchases are enabled.

• Select Manage auto-reload if that control appears on the Billing page.
• Select Turn off.
• Confirm that auto-reload is disabled.

Use the no-credit path when needed

Skip the live extraction if the credit balance is empty. Skip it as well if you cannot set the organization spend limit.

You can still complete the extraction code. The deterministic quote test in the next step does not call Claude or spend API credits.

Build the extraction contract

Pydantic gives the extraction a typed shape. Quote validation then checks the model output against the exact source text before anything reaches the database.

• In newsroom.py, extend the import section before loading the local environment with these lines:

from typing import Literal

from anthropic import Anthropic
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

What do these imports enable?

• The Anthropic Python SDK creates the Claude client used for the extraction request.
• The python-dotenv loader places the ignored local key into the process environment.
• The BaseModel class defines the validated response shape.
• The Literal type restricts entity categories to the two newsroom-relevant values.

• Save newsroom.py.

Your editor should recognize the four imported names without unresolved-reference warnings.

Seeing unresolved imports?

Confirm that the imports sit at the top of newsroom.py. Check that your editor still uses the uv-managed environment from this project.

If one dependency remains unresolved, help me inspect the uv environment.

• Below COMPANIES_HOUSE_API_ROOT in newsroom.py, add the model identifier plus the extraction models:

CLAUDE_MODEL = "claude-haiku-5-5"


class EntityMention(BaseModel):
    entity_text: str
    entity_type: Literal["PERSON", "COMPANY"]
    source_quote: str


class EntityExtraction(BaseModel):
    entities: list[EntityMention]

What does the contract require?

• The CLAUDE_MODEL constant records the exact model used for every stored result.
• The entity_text field holds the extracted name.
• The entity_type field accepts only PERSON or COMPANY.
• The source_quote field carries the evidence claimed by the model.
• The EntityExtraction wrapper gives the structured output one predictable list.

• Save newsroom.py.

Your editor should now recognize both model classes. It should also accept the two literal entity categories.

Seeing a type or indentation warning?

Place both classes at the left edge of the file. Confirm that Literal appears in the import section.

For a warning that remains, help me compare the extraction models.

A valid schema proves that the response has the expected fields. A second check must prove that each quote exists inside the stored source text.

• Below ingest_live_data() in newsroom.py, add the quote validator plus the contract-selection helper:

def validate_source_quote(source_text: str, source_quote: str) -> None:
    if not source_quote or source_quote not in source_text:
        raise ValueError("Extracted source_quote is not a verbatim source substring.")


def find_focus_contract(con: duckdb.DuckDBPyConnection) -> tuple:
    contracts = con.execute(
        """
        SELECT source_id, title, description, buyer, supplier, source_url
        FROM contracts
        WHERE source_name = ?
        ORDER BY source_id
        """,
        ["Contracts Finder"],
    ).fetchall()
    companies = con.execute(
        "SELECT company_number, company_name FROM companies WHERE company_number <> ?",
        ["SYN-001"],
    ).fetchall()

    for contract in contracts:
        supplier_key = normalise_name(contract[4])
        if any(normalise_name(company[1]) == supplier_key for company in companies):
            return contract
    if contracts:
        return contracts[0]
    raise RuntimeError("Run the ingest command before extraction.")

How do these helpers protect the evidence?

• The validate_source_quote() function accepts only a non-empty literal substring from the source.
• The find_focus_contract() function prefers a live supplier that already matches a stored company.
• The fallback keeps extraction available when the small live batch has no exact normalized company match.
• The final error tells the learner that live ingestion must happen before extraction.

• Save newsroom.py.

The new helpers should sit between ingest_live_data() and the extraction function you add next.

Helper code in the wrong location?

Confirm that neither helper is nested inside ingest_live_data(). Both def lines must start at the left edge.

If the structure is unclear, help me place the quote-validation helpers.

How should the extraction flow work?

• The selected contract contributes its title, description, buyer, supplier, source ID, and source URL.
• The source_text value labels each stored field before sending it to Claude.
• The prompt requests named people plus companies only. It also requests the shortest exact source substring.
• The parsed response must match EntityExtraction.
• Every quote passes through validate_source_quote() before storage.
• Each company name passes through normalise_name() before comparison with stored company records.
• Each database row preserves provenance through its source text, source ID, quote, match status, company number, and model ID.

• Define extract_and_store_entities() below find_focus_contract() using the extraction flow above.
• Add the extract subcommand to main() so it prints the stored results as formatted JSON.

✔️ Awesome, I've got everything!

Great. Save newsroom.py before moving to the live-call decision.

ⓧ I'd like to double check the full code

Compare your cumulative newsroom.py file with this reference. Correct every difference before continuing:

from __future__ import annotations

import argparse
import base64
import getpass
import json
import re
from typing import Literal
from urllib.parse import urlencode
from urllib.request import Request, urlopen

import duckdb
from anthropic import Anthropic
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

DB_PATH = "newsroom.duckdb"
CONTRACTS_FINDER_URL = (
    "https://www.contractsfinder.service.gov.uk/Published/Notices/OCDS/Search"
)
COMPANIES_HOUSE_URL = (
    "https://api.company-information.service.gov.uk/search/companies"
)
COMPANIES_HOUSE_API_ROOT = "https://api.company-information.service.gov.uk"
CLAUDE_MODEL = "claude-haiku-5-5"


class EntityMention(BaseModel):
    entity_text: str
    entity_type: Literal["PERSON", "COMPANY"]
    source_quote: str


class EntityExtraction(BaseModel):
    entities: list[EntityMention]


def connect(db_path: str = DB_PATH) -> duckdb.DuckDBPyConnection:
    return duckdb.connect(db_path)


def init_db(con: duckdb.DuckDBPyConnection) -> None:
    con.execute(
        """
        CREATE TABLE IF NOT EXISTS contracts (
            source_id VARCHAR,
            title VARCHAR,
            description VARCHAR,
            buyer VARCHAR,
            supplier VARCHAR,
            value_gbp DOUBLE,
            source_url VARCHAR,
            source_name VARCHAR
        )
        """
    )
    con.execute(
        """
        CREATE TABLE IF NOT EXISTS companies (
            company_number VARCHAR,
            company_name VARCHAR,
            company_status VARCHAR,
            address VARCHAR,
            source_url VARCHAR
        )
        """
    )
    con.execute(
        """
        CREATE TABLE IF NOT EXISTS extracted_entities (
            source_id VARCHAR,
            entity_text VARCHAR,
            entity_type VARCHAR,
            source_quote VARCHAR,
            source_text VARCHAR,
            matched_company_number VARCHAR,
            match_status VARCHAR,
            model_id VARCHAR
        )
        """
    )


def normalise_name(value: str) -> str:
    cleaned = re.sub(r"[^A-Z0-9]+", " ", value.upper()).strip()
    tokens = cleaned.split()
    suffixes = {"LTD", "LIMITED", "LLP", "PLC"}
    while tokens and tokens[-1] in suffixes:
        tokens.pop()
    return " ".join(tokens)


def load_synthetic(con: duckdb.DuckDBPyConnection) -> None:
    con.execute("DELETE FROM contracts WHERE source_id = ?", ["synthetic-contract-001"])
    con.execute("DELETE FROM companies WHERE company_number = ?", ["SYN-001"])
    con.execute(
        "INSERT INTO contracts VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
        [
            "synthetic-contract-001",
            "Council analytics review",
            "Reporter note: Maya Chen says Northstar Analytics won the analytics work.",
            "Riverton Council",
            "Northstar Analytics",
            125000.0,
            "local fixture",
            "Synthetic fixture",
        ],
    )
    con.execute(
        "INSERT INTO companies VALUES (?, ?, ?, ?, ?)",
        [
            "SYN-001",
            "NORTHSTAR ANALYTICS LTD",
            "active",
            "1 Civic Way, Riverton",
            "local fixture",
        ],
    )


def count_exact_synthetic_links(con: duckdb.DuckDBPyConnection) -> int:
    row = con.execute(
        """
        SELECT COUNT(*)
        FROM contracts AS c
        JOIN companies AS co ON c.supplier = co.company_name
        WHERE c.source_id = ?
        """,
        ["synthetic-contract-001"],
    ).fetchone()
    return int(row[0]) if row else 0


def count_normalised_synthetic_links(con: duckdb.DuckDBPyConnection) -> int:
    supplier_row = con.execute(
        "SELECT supplier FROM contracts WHERE source_id = ?",
        ["synthetic-contract-001"],
    ).fetchone()
    company_rows = con.execute(
        "SELECT company_name FROM companies WHERE company_number = ?",
        ["SYN-001"],
    ).fetchall()
    if not supplier_row:
        return 0
    supplier_key = normalise_name(supplier_row[0])
    return sum(normalise_name(row[0]) == supplier_key for row in company_rows)


def fetch_json(url: str, headers: dict[str, str] | None = None) -> dict:
    request = Request(url, headers=headers or {"Accept": "application/json"})
    with urlopen(request, timeout=30) as response:
        return json.load(response)


def first_document_url(documents: list[dict]) -> str:
    for document in documents:
        url = document.get("url", "")
        if url:
            return str(url)
    return ""


def fetch_contracts_finder_awards(limit: int = 5) -> list[dict[str, object]]:
    query = urlencode({"stages": "award", "limit": str(limit)})
    payload = fetch_json(f"{CONTRACTS_FINDER_URL}?{query}")
    records: list[dict[str, object]] = []

    for release in payload.get("releases", []):
        tender = release.get("tender", {}) or {}
        awards = release.get("awards", []) or []
        award = awards[0] if awards else {}
        suppliers = award.get("suppliers", []) or []
        supplier = str(suppliers[0].get("name", "")).strip() if suppliers else ""
        if not supplier:
            continue

        value = award.get("value", {}) or tender.get("value", {}) or {}
        source_url = first_document_url(award.get("documents", []) or [])
        if not source_url:
            source_url = first_document_url(tender.get("documents", []) or [])
        if not source_url:
            source_url = str(payload.get("uri", CONTRACTS_FINDER_URL))

        records.append(
            {
                "source_id": str(release.get("id") or release.get("ocid") or ""),
                "title": str(tender.get("title", "")),
                "description": str(tender.get("description", "")),
                "buyer": str((release.get("buyer", {}) or {}).get("name", "")),
                "supplier": supplier,
                "value_gbp": float(value.get("amount", 0) or 0),
                "source_url": source_url,
                "source_name": "Contracts Finder",
            }
        )

    if not records:
        raise RuntimeError("Contracts Finder returned no award records with suppliers.")
    return records


def save_contracts(con: duckdb.DuckDBPyConnection, records: list[dict[str, object]]) -> None:
    for record in records:
        con.execute("DELETE FROM contracts WHERE source_id = ?", [record["source_id"]])
        con.execute(
            "INSERT INTO contracts VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            [
                record["source_id"],
                record["title"],
                record["description"],
                record["buyer"],
                record["supplier"],
                record["value_gbp"],
                record["source_url"],
                record["source_name"],
            ],
        )


def fetch_companies_house_matches(
    query: str, api_key: str, items_per_page: int = 5
) -> list[dict[str, str]]:
    credentials = base64.b64encode(f"{api_key}:".encode()).decode()
    params = urlencode({"q": query, "items_per_page": str(items_per_page)})
    payload = fetch_json(
        f"{COMPANIES_HOUSE_URL}?{params}",
        {
            "Accept": "application/json",
            "Authorization": f"Basic {credentials}",
        },
    )
    matches: list[dict[str, str]] = []
    for item in payload.get("items", []):
        company_number = str(item.get("company_number", ""))
        relative_url = str((item.get("links", {}) or {}).get("self", ""))
        matches.append(
            {
                "company_number": company_number,
                "company_name": str(item.get("title", "")),
                "company_status": str(item.get("company_status", "")),
                "address": str(item.get("address_snippet", "")),
                "source_url": (
                    f"{COMPANIES_HOUSE_API_ROOT}{relative_url}"
                    if relative_url
                    else f"{COMPANIES_HOUSE_URL}?{params}"
                ),
            }
        )
    return matches


def save_companies(con: duckdb.DuckDBPyConnection, matches: list[dict[str, str]]) -> None:
    for match in matches:
        con.execute(
            "DELETE FROM companies WHERE company_number = ?",
            [match["company_number"]],
        )
        con.execute(
            "INSERT INTO companies VALUES (?, ?, ?, ?, ?)",
            [
                match["company_number"],
                match["company_name"],
                match["company_status"],
                match["address"],
                match["source_url"],
            ],
        )


def ingest_live_data(con: duckdb.DuckDBPyConnection) -> tuple[int, int, str]:
    contracts = fetch_contracts_finder_awards()
    save_contracts(con, contracts)

    api_key = getpass.getpass("Companies House API key: ").strip()
    if not api_key:
        raise RuntimeError("A Companies House API key is required for enrichment.")

    company_count = 0
    matched_supplier = ""
    for contract in contracts:
        matches = fetch_companies_house_matches(str(contract["supplier"]), api_key)
        if matches:
            save_companies(con, matches)
            company_count = len(matches)
            matched_supplier = str(contract["supplier"])
            break

    if not matched_supplier:
        raise RuntimeError("No Companies House match was found for the fetched suppliers.")
    return len(contracts), company_count, matched_supplier


def validate_source_quote(source_text: str, source_quote: str) -> None:
    if not source_quote or source_quote not in source_text:
        raise ValueError("Extracted source_quote is not a verb source substring.")


def find_focus_contract(con: duckdb.DuckDBPyConnection) -> tuple:
    contracts = con.execute(
        """
        SELECT source_id, title, description, buyer, supplier, source_url
        FROM contracts
        WHERE source_name = ?
        ORDER BY source_id
        """,
        ["Contracts Finder"],
    ).fetchall()
    companies = con.execute(
        "SELECT company_number, company_name FROM companies WHERE company_number <> ?",
        ["SYN-001"],
    ).fetchall()

    for contract in contracts:
        supplier_key = normalise_name(contract[4])
        if any(normalise_name(company[1]) == supplier_key for company in companies):
            return contract
    if contracts:
        return contracts[0]
    raise RuntimeError("Run the ingest command before extraction.")


def extract_and_store_entities(con: duckdb.DuckDBPyConnection) -> list[dict[str, str]]:
    source_id, title, description, buyer, supplier, source_url = find_focus_contract(con)
    source_text = (
        f"Title: {title}\n"
        f"Description: {description}\n"
        f"Buyer: {buyer}\n"
        f"Awarded supplier: {supplier}"
    )

    client = Anthropic()
    response = client.messages.parse(
        model=CLAUDE_MODEL,
        max_tokens=600,
        messages=[
            {
                "role": "user",
                "content": (
                    "Extract named people and companies from the source text. "
                    "For source_quote, copy the shortest exact substring that proves "
                    "the entity appears. Do not infer an entity that is absent.\n\n"
                    f"{source_text}"
                ),
            }
        ],
        output_format=EntityExtraction,
    )
    parsed = response.parsed_output
    if parsed is None:
        raise RuntimeError("Claude did not return a parsed extraction.")

    companies = con.execute(
        "SELECT company_number, company_name FROM companies"
    ).fetchall()
    con.execute("DELETE FROM extracted_entities WHERE source_id = ?", [source_id])

    stored: list[dict[str, str]] = []
    for entity in parsed.entities:
        validate_source_quote(source_text, entity.source_quote)
        matched_number = ""
        if entity.entity_type == "COMPANY":
            entity_key = normalise_name(entity.entity_text)
            for company_number, company_name in companies:
                if normalise_name(company_name) == entity_key:
                    matched_number = str(company_number)
                    break
        match_status = "linked" if matched_number else "unlinked"
        con.execute(
            "INSERT INTO extracted_entities VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            [
                source_id,
                entity.entity_text,
                entity.entity_type,
                entity.source_quote,
                source_text,
                matched_number,
                match_status,
                CLAUDE_MODEL,
            ],
        )
        stored.append(
            {
                "entity_text": entity.entity_text,
                "entity_type": entity.entity_type,
                "source_quote": entity.source_quote,
                "match_status": match_status,
                "matched_company_number": matched_number,
                "source_id": str(source_id),
                "source_url": str(source_url),
                "model_id": CLAUDE_MODEL,
            }
        )
    return stored


def main() -> None:
    parser = argparse.ArgumentParser(description="Newsroom Evidence Finder")
    subparsers = parser.add_subparsers(dest="command", required=True)
    subparsers.add_parser("synthetic")
    subparsers.add_parser("ingest")
    subparsers.add_parser("extract")
    args = parser.parse_args()

    with connect() as con:
        init_db(con)
        if args.command == "synthetic":
            load_synthetic(con)
            print(f"Exact-name links: {count_exact_synthetic_links(con)}")
        elif args.command == "ingest":
            load_synthetic(con)
            print(
                f"Normalised synthetic links: {count_normalised_synthetic_links(con)}"
            )
            contract_count, company_count, supplier = ingest_live_data(con)
            print(f"Contracts Finder records: {contract_count}")
            print(f"Companies House matches: {company_count}")
            print(f"Enriched supplier: {supplier}")
        elif args.command == "extract":
            print(json.dumps(extract_and_store_entities(con), indent=2))


if __name__ == "__main__":
    main()

What does the complete file include?

This reference keeps every synthetic-data function plus the live-ingestion functions from earlier steps. It adds the models, quote validator, contract selector, extraction workflow, and extract command required in this step.

Run and document the evidence chain

The live path now has three safeguards. Your balance controls whether you run it. The typed contract controls the response shape.

The literal-substring check controls what reaches the database. Choose the path that matches the balance you saw in Claude Console.

✔️ I have credits available

Before you run the extraction, which provenance fields do you expect to see beside each entity?

• Run the live extraction from PowerShell with this command:

uv run python newsroom.py extract

What should I see?

• Each item identifies a PERSON or COMPANY.
• Each source_quote appears literally inside the stored source text.
• The source_id identifies the imported contract record.
• The source_url points back to the public evidence.
• The match_status reports linked or unlinked.
• The model_id reports claude-haiku-5-5.

That is the evidence chain working. Every accepted entity now carries enough context for a reporter to verify it.

• Return to Billing in Claude Console.
• Check the credit usage recorded for this extraction call.

Extraction did not complete?

Confirm that your ignored .env contains the Claude key from setup. Check that live Contracts Finder records remain in newsroom.duckdb.

If the request or parsed output still fails, help me debug the traceable extraction.

ⓧ I have no credits

Skipping the live call protects your account from an unapproved charge. Your completed extraction code remains ready for a future run.

• Leave the live extraction command unrun.
• Keep the completed newsroom.py file for the deterministic tests in the next step.

Your project still demonstrates the evidence contract through its typed models, literal-quote validator, provenance fields, and testable business logic.

• Replace the second paragraph in README.md with this sentence: Claude structured extraction is optional when credits are unavailable. When used, every entity must retain a literal source quote, source record, company-link status, and model ID.
• Add one personal Step 4 | Did: ... | Broke: ... | Learned: ... entry to NOTES.md.

Explain aloud why a valid schema cannot prove that a quote came from the source. Use your own words before saving the reflection.

Your extraction stage now treats model output as a claim that must carry evidence. Next, you will expose that evidence through MCP tools before proving the pipeline with deterministic tests.

Expose MCP Tools and Publish the Demo

Your pipeline now turns public records into evidence-backed entities. Reporters still need focused ways to search those records or trace a name through its sources. This is where the Model Context Protocol turns your pipeline into callable newsroom tools.

Interviewers also need proof that the underlying logic works without live services. You will add deterministic tests before publishing the complete demo through GitHub.

In this step, get ready to:

• Expose evidence search and entity tracing through MCP tools.
• Prove the pipeline with deterministic tests and MCP Inspector.
• Document the design before publishing a credential-free repository.

Add searchable provenance and MCP tools

The reporter-facing functions keep business logic inside newsroom.py. The adapter in server.py stays small because it only exposes those functions through MCP.

• In newsroom.py, add search_records() directly above main() using this code:

def search_records(query: str, db_path: str = DB_PATH) -> list[dict[str, object]]:
    with connect(db_path) as con:
        init_db(con)
        pattern = f"%{query}%"
        rows = con.execute(
            """
            SELECT source_id, title, buyer, supplier, value_gbp, source_url, source_name
            FROM contracts
            WHERE lower(title || ' ' || description || ' ' || buyer || ' ' || supplier)
                  LIKE lower(?)
            ORDER BY value_gbp DESC
            LIMIT 10
            """,
            [pattern],
        ).fetchall()
    return [
        {
            "source_id": row[0],
            "title": row[1],
            "buyer": row[2],
            "supplier": row[3],
            "value_gbp": row[4],
            "source_url": row[5],
            "source_name": row[6],
        }
        for row in rows
    ]

What does this search preserve?

• The search checks each contract title.
• The search also checks descriptions.
• Buyer names and supplier names remain searchable.
• Every result retains its source identifier.
• Every result also retains its public source reference.

• Add the query section of trace_entity() directly below search_records() using this code:

def trace_entity(name: str, db_path: str = DB_PATH) -> dict[str, object]:
    pattern = f"%{name}%"
    with connect(db_path) as con:
        init_db(con)
        extractions = con.execute(
            """
            SELECT e.entity_text, e.entity_type, e.source_quote, e.match_status,
                   e.matched_company_number, e.model_id, c.source_id, c.source_url
            FROM extracted_entities AS e
            JOIN contracts AS c ON e.source_id = c.source_id
            WHERE lower(e.entity_text) LIKE lower(?)
            """,
            [pattern],
        ).fetchall()
        companies = con.execute(
            """
            SELECT company_number, company_name, company_status, address, source_url
            FROM companies
            WHERE lower(company_name) LIKE lower(?)
            """,
            [pattern],
        ).fetchall()

What do these queries collect?

The first query joins each matching extraction back to its contract. This preserves the exact quote beside its source record.

The second query finds matching Companies House records. The final response can therefore show extraction evidence beside company evidence.

• Continue trace_entity() with the extraction section of its return value:

    return {
        "query": name,
        "extractions": [
            {
                "entity_text": row[0],
                "entity_type": row[1],
                "source_quote": row[2],
                "match_status": row[3],
                "matched_company_number": row[4],
                "model_id": row[5],
                "source_id": row[6],
                "source_url": row[7],
            }
            for row in extractions
        ],

Why return the extraction metadata?

The extraction result keeps the model ID beside the evidence quote. It also carries the company-link status and source reference needed for editorial review.

• Complete the same return value with company records and related contracts:

        "companies": [
            {
                "company_number": row[0],
                "company_name": row[1],
                "company_status": row[2],
                "address": row[3],
                "source_url": row[4],
            }
            for row in companies
        ],
        "contracts": search_records(name, db_path),
    }

How does the trace come together?

The completed response combines extraction evidence with Companies House details. It also calls search_records() to include related contracts.

• In main(), add the search and trace parsers directly below the extract parser:

    search_parser = subparsers.add_parser("search")
    search_parser.add_argument("query")
    trace_parser = subparsers.add_parser("trace")
    trace_parser.add_argument("name")

What do these parsers add?

The two parsers give the local application searchable commands. Each parser captures the value passed by the reporter.

• Below args = parser.parse_args(), add the command dispatch before the existing database connection:

    if args.command == "search":
        print(json.dumps(search_records(args.query), indent=2))
        return
    if args.command == "trace":
        print(json.dumps(trace_entity(args.name), indent=2))
        return

Why return after each command?

Each branch prints one JSON response before leaving main(). This prevents the synthetic or live-ingestion branches from running afterward.

• Save newsroom.py.
• Replace the placeholder content in server.py with this MCP adapter:

import json

from mcp.server import MCPServer

from newsroom import search_records as search_records_db
from newsroom import trace_entity as trace_entity_db

mcp = MCPServer("Newsroom Evidence Finder")


@mcp.tool()
def search_records(query: str) -> str:
    """Search contracts and return evidence with source references."""
    return json.dumps(search_records_db(query), indent=2)


@mcp.tool()
def trace_entity(name: str) -> str:
    """Trace an entity to extracted quotes, company matches, and contracts."""
    return json.dumps(trace_entity_db(name), indent=2)


if __name__ == "__main__":
    mcp.run()

What does the adapter do?

• MCPServer creates a server named Newsroom Evidence Finder.
• Each @mcp.tool() decorator turns one Python function into a callable tool.
• The adapter delegates database work to newsroom.py.
• JSON strings keep the evidence easy to inspect in the tool result.

• Save server.py.

Seeing a Python syntax error?

Check that every section of trace_entity() has the same four-space indentation. Confirm that the final closing brace appears before main().

Compare the two imported aliases in server.py with the function names in newsroom.py.

Ask for help with the exact traceback: help me debug my newsroom MCP adapter

✔️ Awesome, I've got everything!

Great. Save server.py before moving to the deterministic tests.

ⓧ I'd like to double check the full code

import json

from mcp.server import MCPServer

from newsroom import search_records as search_records_db
from newsroom import trace_entity as trace_entity_db

mcp = MCPServer("Newsroom Evidence Finder")


@mcp.tool()
def search_records(query: str) -> str:
    """Search contracts and return evidence with source references."""
    return json.dumps(search_records_db(query), indent=2)


@mcp.tool()
def trace_entity(name: str) -> str:
    """Trace an entity to extracted quotes, company matches, and contracts."""
    return json.dumps(trace_entity_db(name), indent=2)


if __name__ == "__main__":
    mcp.run()

Test the pipeline in MCP Inspector

Live records can change between runs. Network access can also fail. Deterministic tests use a temporary DuckDB database so those external conditions cannot decide whether your test suite passes.

• Replace the placeholder content in test_newsroom.py with the imports and test fixture below:

import tempfile
import unittest
from pathlib import Path

from newsroom import (
    connect,
    count_exact_synthetic_links,
    count_normalised_synthetic_links,
    init_db,
    load_synthetic,
    normalise_name,
    search_records,
    validate_source_quote,
)

Why use a temporary database?

Each test run gets an isolated database path. Your live newsroom.duckdb file remains untouched.

• Add the test class below the imports:

class EvidenceFinderTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temp_dir = tempfile.TemporaryDirectory()
        self.db_path = str(Path(self.temp_dir.name) / "test.duckdb")
        self.con = connect(self.db_path)
        init_db(self.con)
        load_synthetic(self.con)

    def tearDown(self) -> None:
        self.con.close()
        self.temp_dir.cleanup()

    def test_exact_join_exposes_alias_problem(self) -> None:
        self.assertEqual(count_exact_synthetic_links(self.con), 0)

    def test_normalisation_resolves_company_suffix(self) -> None:
        self.assertEqual(normalise_name("Northstar Analytics Ltd"), "NORTHSTAR ANALYTICS")
        self.assertEqual(count_normalised_synthetic_links(self.con), 1)

What do the first tests prove?

The first assertion preserves the designed exact-name failure. The second proves that normalization removes the company suffix before matching.

• Complete the class with source-preservation and quote-validation tests:

    def test_search_preserves_source_reference(self) -> None:
        results = search_records("analytics", self.db_path)
        self.assertEqual(results[0]["source_id"], "synthetic-contract-001")
        self.assertTrue(results[0]["source_url"])

    def test_source_quote_must_be_verbatim(self) -> None:
        source = "Maya Chen joined Northstar Analytics."
        validate_source_quote(source, "Maya Chen")
        with self.assertRaises(ValueError):
            validate_source_quote(source, "Maya Chen founded Northstar Analytics")


if __name__ == "__main__":
    unittest.main()

Why do these tests matter?

The search test proves that a useful result still carries provenance. The quote test proves that invented evidence is rejected before storage.

No test calls Contracts Finder or Companies House. No test calls Claude or spends API credits.

• Save test_newsroom.py.

Before you run the suite, which result would prove all four checks passed without a live API?

• Run the deterministic test suite in PowerShell:

uv run python -m unittest -v

What does this command check?

uv runs Python inside the pinned project environment. The unittest runner discovers the local test module before reporting each result.

You should see all four tests pass. The final line ends with OK.

Tests not ending with OK?

If the import fails, compare every imported function name with newsroom.py. If a source test fails, confirm search_records() returns both source_id and source_url.

Use the first failing traceback as your starting point: help me diagnose the failing deterministic test

✔️ Awesome, I've got everything!

That is the core proof complete. Your pipeline passes without network access or API spending.

ⓧ I'd like to double check the full code

import tempfile
import unittest
from pathlib import Path

from newsroom import (
    connect,
    count_exact_synthetic_links,
    count_normalised_synthetic_links,
    init_db,
    load_synthetic,
    normalise_name,
    search_records,
    validate_source_quote,
)


class EvidenceFinderTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temp_dir = tempfile.TemporaryDirectory()
        self.db_path = str(Path(self.temp_dir.name) / "test.duckdb")
        self.con = connect(self.db_path)
        init_db(self.con)
        load_synthetic(self.con)

    def tearDown(self) -> None:
        self.con.close()
        self.temp_dir.cleanup()

    def test_exact_join_exposes_alias_problem(self) -> None:
        self.assertEqual(count_exact_synthetic_links(self.con), 0)

    def test_normalisation_resolves_company_suffix(self) -> None:
        self.assertEqual(normalise_name("Northstar Analytics Ltd"), "NORTHSTAR ANALYTICS")
        self.assertEqual(count_normalised_synthetic_links(self.con), 1)

    def test_search_preserves_source_reference(self) -> None:
        results = search_records("analytics", self.db_path)
        self.assertEqual(results[0]["source_id"], "synthetic-contract-001")
        self.assertTrue(results[0]["source_url"])

    def test_source_quote_must_be_verbatim(self) -> None:
        source = "Maya Chen joined Northstar Analytics."
        validate_source_quote(source, "Maya Chen")
        with self.assertRaises(ValueError):
            validate_source_quote(source, "Maya Chen founded Northstar Analytics")


if __name__ == "__main__":
    unittest.main()

MCP Inspector lets you call each tool without building a separate client. Its result panel makes the preserved source records visible.

• Launch the MCP server in MCP Inspector by running:

uv run mcp dev server.py

What does this command start?

The MCP development command starts server.py through the project environment. It prints the local MCP Inspector URL in PowerShell.

Keep this PowerShell process running while you use the Inspector.

The first launch can take longer while MCP Inspector starts. A short wait here does not mean the server has failed.

• Open the printed MCP Inspector URL in your browser.
• Select the Tools tab.
• Select search_records from the available tools.
• Enter analytics for the query.

Before you call the tool, do you expect the result to keep the synthetic source reference?

• Call search_records.

You should see the synthetic contract in the result. Its fields include synthetic-contract-001 and a non-empty source reference.

• Select trace_entity from the available tools.
• Enter Northstar Analytics for the name.
• Call trace_entity.

You should see related contract and company records with source references. If you completed the optional live Claude call earlier, matching extraction entries also show their exact quotes and model IDs.

Tools missing from MCP Inspector?

Confirm PowerShell is running from the newsroom-evidence-finder folder. Check that server.py imports both database functions and decorates both adapter functions.

If the server process stopped, use its traceback to identify the first import or syntax failure: help me restore my MCP tools

Document and publish the repository

A public repository should explain the newsroom problem before showing commands. It should also make the evidence model and test strategy clear to another developer.

• Replace the current content in README.md with this first section:

# Newsroom Evidence Finder

A local Python pipeline that helps investigative reporters search UK public-contract records, reconcile company names, and trace AI-extracted entities back to evidence.

## What it demonstrates

- Synthetic and live ETL into DuckDB
- Contracts Finder OCDS ingestion
- Companies House supplier enrichment
- Company-name normalization and entity resolution
- Claude structured extraction with verbatim quote validation
- MCP search and traceability tools
- Deterministic tests that do not call external APIs

## Architecture

Contracts Finder and Companies House feed normalized DuckDB tables. Claude receives a source-backed text assembled from one imported record. The pipeline stores only extractions whose quote occurs literally in that text. MCP tools read the database but do not bypass the provenance fields.

## Demo

```powershell
uv run python newsroom.py synthetic
uv run python newsroom.py ingest
uv run python newsroom.py extract
uv run python -m unittest -v
uv run mcp dev server.py
```

What does the opening establish?

The feature list shows the complete workflow. The architecture section explains how public records become searchable evidence without losing provenance.

• Complete README.md with the design decisions and evidence sections:

## Design decisions

- DuckDB keeps analytical tables and joins local, inspectable, and reproducible.
- Evidence-first means every extracted entity keeps an exact quote and public source reference.
- Business logic stays outside the MCP adapter so it can be tested without starting a server.
- A follow-on project can move raw files to Amazon S3 and provision infrastructure with Terraform.

## Evidence gallery

Add missed-link, resolved-link, passing-test, and MCP Inspector screenshots here. Do not include credentials.

## Limitations and next steps

This portfolio version processes a deliberately small batch. Live public records can change between runs.

Why include limitations?

The final section states that live public data can change. This makes the demo honest about reproducibility without weakening the deterministic test evidence.

• Save README.md.
• Add the missed-link screenshot to the evidence gallery.
• Add the resolved-link screenshot to the evidence gallery.
• Add the passing-test screenshot to the evidence gallery.
• Add the MCP Inspector screenshot to the evidence gallery.

✔️ Awesome, I've got everything!

Your README now explains the problem, architecture, evidence controls, tests, limitations, and follow-on direction.

ⓧ I'd like to double check the full code

# Newsroom Evidence Finder

A local Python pipeline that helps investigative reporters search UK public-contract records, reconcile company names, and trace AI-extracted entities back to evidence.

## What it demonstrates

- Synthetic and live ETL into DuckDB
- Contracts Finder OCDS ingestion
- Companies House supplier enrichment
- Company-name normalization and entity resolution
- Claude structured extraction with verbatim quote validation
- MCP search and traceability tools
- Deterministic tests that do not call external APIs

## Architecture

Contracts Finder and Companies House feed normalized DuckDB tables. Claude receives a source-backed text assembled from one imported record. The pipeline stores only extractions whose quote occurs literally in that text. MCP tools read the database but do not bypass the provenance fields.

## Demo

```powershell
uv run python newsroom.py synthetic
uv run python newsroom.py ingest
uv run python newsroom.py extract
uv run python -m unittest -v
uv run mcp dev server.py
```

## Design decisions

- DuckDB keeps analytical tables and joins local, inspectable, and reproducible.
- Evidence-first means every extracted entity keeps an exact quote and public source reference.
- Business logic stays outside the MCP adapter so it can be tested without starting a server.
- A follow-on project can move raw files to Amazon S3 and provision infrastructure with Terraform.

## Evidence gallery

Add missed-link, resolved-link, passing-test, and MCP Inspector screenshots here. Do not include credentials.

## Limitations and next steps

This portfolio version processes a deliberately small batch. Live public records can change between runs.

• In NOTES.md, add one Step 5 line in your own words.
• Explain the separation between business logic and the MCP adapter in that entry.
• Record one test or Inspector issue you investigated in that entry.

Publishing makes the repository public. Take a careful credential check now so your local Claude key stays on your computer.

• Check the repository state by running:

git status

What should the status prove?

The status output should list your code, tests, README, notes, and screenshots. It must not list .env.

If .env appears, stop before staging anything. Confirm the ignore rules still match that filename.

• Stage the credential-free project and create the final commit by running:

git add .
git commit -m "Build evidence-first newsroom data pipeline"

What does this commit capture?

The commit records the evidence pipeline, MCP adapter, deterministic tests, documentation, screenshots, and learning log. The ignored local secrets and database remain outside the commit.

Commit blocked or showing the wrong files?

If Git reports an identity problem, follow the guidance printed in PowerShell before retrying the commit. If a credential file is staged, remove it from the staged changes before continuing.

Ask for a credential-safe diagnosis before pushing: help me inspect my staged Git files

• On GitHub, create a new public repository named newsroom-evidence-finder.
• Leave the generated README option unselected.
• Leave the license option unselected.
• Leave the generated .gitignore option unselected.
• Create the empty repository.
• Copy the Quick Setup remote URL.

• Connect the local repository by replacing REMOTE-URL with the copied Quick Setup URL before running:

git remote add origin REMOTE-URL

What does the remote add?

The remote names your GitHub repository origin. Your local commits now have a destination for publishing.

• Verify the saved remote by running:

git remote -v

What should the remote show?

You should see the copied repository URL listed for fetch and push operations. Confirm that both entries point to newsroom-evidence-finder.

Before the final push, do you expect the public repository to include .env or newsroom.duckdb?

• Publish the main branch by running:

git push -u origin main

What does the push establish?

The push publishes the committed main branch. It also records origin/main as the upstream branch for later pushes.

• Refresh the public repository page on GitHub.
• Confirm the repository shows newsroom.py.
• Confirm the repository shows server.py.
• Confirm the repository shows test_newsroom.py.
• Confirm the README displays the evidence gallery and design decisions.
• Confirm .env is absent from the repository.

That is the complete demo published. Your repository now proves the evidence workflow through code, tests, source-backed MCP results, and your own engineering notes.

Secret mission

Build Your Own Editorial Quality Gate

Your evidence finder preserves provenance, but incomplete records can still reach a reporter. Design a deterministic quality gate that identifies missing editorial evidence, proves its behavior with local tests, and reports every affected source record.

Clean Up Your Resources

Clean Up Your Resources

The local project has no ongoing costs. Choose whether to keep the demo ready, pause its processes, or remove its resources.

Cost warning

Billing controls can feel high-stakes, but this check keeps the next live call within a limit you choose. Future live extractions can consume Claude API credits.

• Open Settings in Claude Console.
• Select Billing.
• Check Credit balance before another live extraction.
• Confirm auto-reload remains off.
• Keep a small organization limit under Spend limits.

Resources you used:

• Local newsroom-evidence-finder folder containing .venv, newsroom.duckdb, the ignored .env, code, tests, documentation, screenshots, and NOTES.md.
• Local MCP Inspector development process, if it is still running.
• Claude Console API key used for optional extraction.
• Companies House API key used for public company enrichment.
• Public GitHub repository named newsroom-evidence-finder.

Keep everything running

No action needed. Choose this if you are still demonstrating the evidence finder or refining its quality gate.

• Retain the newsroom-evidence-finder folder with its database, screenshots, notes, and ignored credentials.
• Leave the MCP Inspector process running only while you are actively demonstrating the tools.
• Keep the Claude spend limit in place for future live extraction calls.
• Keep the public newsroom-evidence-finder repository available as your remote demo.

Live public records may differ when you run ingestion again.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your local database, ignored credentials, code, screenshots, and notes remain ready for the next session.

• Close the browser tab running MCP Inspector.
• Press Ctrl+C in the PowerShell terminal that started the MCP development server.
• Close the remaining PowerShell windows for this project.
• Leave .env inside the newsroom-evidence-finder folder so it remains protected by .gitignore.
• Recheck the Claude balance and spend controls before the next live extraction.

Delete - I don't want to use this again

Deleting these local files and credentials closes this project on your computer. Your public repository can remain as a remote copy.

• Press Ctrl+C in the PowerShell terminal running the MCP development server.
• Use File Explorer to delete .env from the newsroom-evidence-finder folder.
• Use File Explorer to delete newsroom.duckdb from the newsroom-evidence-finder folder.
• Use File Explorer to delete .venv from the newsroom-evidence-finder folder.
• Delete the remaining newsroom-evidence-finder folder if you no longer want the local code, tests, screenshots, and notes.
• Revoke the project key from the Claude Console API keys page.
• Delete the Companies House API key from the live application's key management page in Developer Hub.
• Delete the public newsroom-evidence-finder repository from its GitHub repository settings only if you no longer want the published demo.

Nice Work!

Nice Work!

You made it! Your evidence-first newsroom pipeline now turns UK public records into traceable reporting leads.

What you learned:

• Build a reproducible DuckDB ETL pipeline from Contracts Finder awards. Enrich supplier candidates through the Companies House Public Data API. Retain source URLs for reporter verification.
• Reveal the exact-name alias failure before resolving it through company-name normalization. Extract named people plus companies with Claude under explicit cost controls. Reject unsupported quotes before storing the result.
• Expose source-backed search plus entity tracing through MCP tools. Demonstrate those tools in MCP Inspector with visible source records. Verify the logic with deterministic tests before publishing an interview-ready GitHub repository.
• Secret Mission: Design an editorial evidence-quality gate that reports missing fields by source record. Prove its behavior with deterministic tests across valid records plus incomplete evidence. Confirm the synthetic database passes the same audit.

Ready to quiz yourself?