add-data-source

Add a New Data Source

Follow these 6 steps in order. Each step must be completed before moving to the next.

Step 1: Create Transport Client

Create property_core/new_client.py. Choose the pattern that fits:

• Sync HTTP API (httpx): see property_core/companies_house_client.py
  • is_configured() check for optional credentials
  • Basic auth or API key from os.getenv()
• Async HTTP API (httpx): see property_core/epc_client.py
• HTML scraper (requests + BeautifulSoup): see property_core/rightmove_scraper.py
• SPARQL/RDF: see property_core/ppd_client.py

Required in all clients:

• from __future__ import annotations
• Type hints on all methods
• Docstring on the class/module
• Rate limiting/retry logic where appropriate

Step 2: Create Pydantic Models

Create property_core/models/new_source.py.

Transport-parsed models get the raw field:

raw: dict | None = Field(default=None, exclude=True)

Use @classmethod factory methods for parsing:

@classmethod
def from_api_response(cls, data: dict) -> "NewModel":
    return cls(field=data.get("key"), raw=data)

Step 3: Export from models/init.py

Open property_core/models/__init__.py:

• Add import: from .new_source import NewModel
• Add to __all__ list

Step 4: Export from property_core/init.py

Open property_core/__init__.py:

• Add client/service import
• Add model import
• Add both to __all__ list

Step 5: Create Domain Service (if needed)

If the data source needs orchestration (combining with other sources, stats computation, guardrails), create property_core/new_service.py.

• Class-based if stateful: class NewService
• Function-based if stateless: def analyze_new()
• Export from property_core/__init__.py

Step 6: Add Consumer Endpoints

All three consumers must be updated:

API Router

Create app/api/v1/new_source.py, register in app/api/routes.py. See /add-endpoint skill for details.

MCP Tool

Add @mcp.tool() function in mcp_server/server.py. See /add-mcp-tool skill for details.