Fetch Full Record from UK Due Diligence Register

Instructions

Implementation Reference

• search_fetch.py:147-227 (handler)

  The `fetch` tool handler function. Routes a prefixed ID (company:, charity:, disqualification:, notice:) to the appropriate backend fetch helper and returns the full record as a dict with id, title, content, and metadata.

  @mcp.tool(
      name="fetch",
      annotations={
          "title": "Fetch Full Record from UK Due Diligence Register",
          "readOnlyHint": True,
          "destructiveHint": False,
          "idempotentHint": True,
          "openWorldHint": True,
      },
  )
  async def fetch(
      id: Annotated[str, Field(description="Prefixed record ID returned by search. Format: company:{number}, charity:{number}, disqualification:{officer_id}, or notice:{notice_id}", min_length=3, max_length=100)],
  ) -> dict:
      """Fetch the full record for an ID returned by search.
  
      Routes by prefix to the appropriate register:
      - company:{number} → Companies House full profile
      - charity:{number} → Charity Commission full profile
      - disqualification:{officer_id} → Disqualified director full record
      - notice:{notice_id} → Gazette notice full legal text
      """
      prefix, _, value = id.partition(":")
      if not value:
          raise ValueError(f"Invalid ID format {id!r} — expected prefix:value")
  
      if prefix == "company":
          co = await _fetch_company_profile(_normalise_company_number(value))
          return {
              "id": id,
              "title": co.company_name or value,
              "content": co.model_dump_json(),
              "metadata": {
                  "source": "companies_house",
                  "status": co.company_status,
                  "company_type": co.company_type,
                  "date_of_creation": co.date_of_creation,
              },
          }
  
      if prefix == "charity":
          ch = await _fetch_charity_profile(value)
          return {
              "id": id,
              "title": ch.charity_name or value,
              "content": ch.model_dump_json(),
              "metadata": {
                  "source": "charity_commission",
                  "status": ch.reg_status_label,
                  "date_of_registration": ch.date_of_registration,
              },
          }
  
      if prefix == "disqualification":
          dq = await _fetch_disqualified_profile(value)
          return {
              "id": id,
              "title": dq.name or value,
              "content": dq.model_dump_json(),
              "metadata": {
                  "source": "companies_house_disqualified",
                  "officer_kind": dq.officer_kind,
              },
          }
  
      if prefix == "notice":
          data: dict[str, Any] = await _fetch_gazette_notice(value)
          title = (
              data.get("title")
              or data.get("f:notice-code")
              or f"Gazette notice {value}"
          )
          return {
              "id": id,
              "title": title,
              "content": json.dumps(data),
              "metadata": {"source": "gazette"},
          }
  
      raise ValueError(
          f"Unknown ID prefix {prefix!r}. Valid prefixes: company, charity, disqualification, notice"
      )

• search_fetch.py:106-227 (registration)

  Registration of both `search` and `fetch` tools via @mcp.tool decorator in the register_tools() function. Server.py calls search_fetch.register_tools(mcp) on line 164.

  def register_tools(mcp: FastMCP) -> None:
  
      @mcp.tool(
          name="search",
          annotations={
              "title": "Search UK Due Diligence Registers",
              "readOnlyHint": True,
              "destructiveHint": False,
              "idempotentHint": True,
              "openWorldHint": True,
          },
      )
      async def search(
          query: Annotated[str, Field(description="Company name, charity name, director name, or keyword to search for across all UK due diligence registers", min_length=2, max_length=200)],
      ) -> dict:
          """Search across all UK due diligence registers simultaneously.
  
          Searches Companies House, Charity Commission, disqualified directors,
          and Gazette insolvency notices in parallel. Returns a list of result
          IDs — use fetch with each ID to retrieve the full record.
          """
          tasks = [
              _company_ids(query),
              _charity_ids(query),
              _disqualified_ids(query),
              _gazette_ids(query),
          ]
          results = await asyncio.gather(*tasks, return_exceptions=True)
  
          ids: list[str] = []
          seen: set[str] = set()
          for result in results:
              if isinstance(result, Exception):
                  continue
              for id_ in result:
                  if id_ not in seen:
                      seen.add(id_)
                      ids.append(id_)
  
          return {"ids": ids}
  
      @mcp.tool(
          name="fetch",
          annotations={
              "title": "Fetch Full Record from UK Due Diligence Register",
              "readOnlyHint": True,
              "destructiveHint": False,
              "idempotentHint": True,
              "openWorldHint": True,
          },
      )
      async def fetch(
          id: Annotated[str, Field(description="Prefixed record ID returned by search. Format: company:{number}, charity:{number}, disqualification:{officer_id}, or notice:{notice_id}", min_length=3, max_length=100)],
      ) -> dict:
          """Fetch the full record for an ID returned by search.
  
          Routes by prefix to the appropriate register:
          - company:{number} → Companies House full profile
          - charity:{number} → Charity Commission full profile
          - disqualification:{officer_id} → Disqualified director full record
          - notice:{notice_id} → Gazette notice full legal text
          """
          prefix, _, value = id.partition(":")
          if not value:
              raise ValueError(f"Invalid ID format {id!r} — expected prefix:value")
  
          if prefix == "company":
              co = await _fetch_company_profile(_normalise_company_number(value))
              return {
                  "id": id,
                  "title": co.company_name or value,
                  "content": co.model_dump_json(),
                  "metadata": {
                      "source": "companies_house",
                      "status": co.company_status,
                      "company_type": co.company_type,
                      "date_of_creation": co.date_of_creation,
                  },
              }
  
          if prefix == "charity":
              ch = await _fetch_charity_profile(value)
              return {
                  "id": id,
                  "title": ch.charity_name or value,
                  "content": ch.model_dump_json(),
                  "metadata": {
                      "source": "charity_commission",
                      "status": ch.reg_status_label,
                      "date_of_registration": ch.date_of_registration,
                  },
              }
  
          if prefix == "disqualification":
              dq = await _fetch_disqualified_profile(value)
              return {
                  "id": id,
                  "title": dq.name or value,
                  "content": dq.model_dump_json(),
                  "metadata": {
                      "source": "companies_house_disqualified",
                      "officer_kind": dq.officer_kind,
                  },
              }
  
          if prefix == "notice":
              data: dict[str, Any] = await _fetch_gazette_notice(value)
              title = (
                  data.get("title")
                  or data.get("f:notice-code")
                  or f"Gazette notice {value}"
              )
              return {
                  "id": id,
                  "title": title,
                  "content": json.dumps(data),
                  "metadata": {"source": "gazette"},
              }
  
          raise ValueError(
              f"Unknown ID prefix {prefix!r}. Valid prefixes: company, charity, disqualification, notice"
          )

• search_fetch.py:157-159 (schema)

  Input schema for the fetch tool: a single 'id' parameter — a prefixed record ID with format company:{number}, charity:{number}, disqualification:{officer_id}, or notice:{notice_id}.

  async def fetch(
      id: Annotated[str, Field(description="Prefixed record ID returned by search. Format: company:{number}, charity:{number}, disqualification:{officer_id}, or notice:{notice_id}", min_length=3, max_length=100)],
  ) -> dict:

• search_fetch.py:8-28 (schema)

  ID scheme documentation and imports for the four backend fetch helpers: _fetch_company_profile, _fetch_charity_profile, _fetch_disqualified_profile, _fetch_gazette_notice.

  ID scheme:
    company:{company_number}          → Companies House profile
    charity:{charity_number}          → Charity Commission profile
    disqualification:{officer_id}     → Disqualified director profile
    notice:{notice_numeric_id}        → Gazette notice full text
  """
  
  from __future__ import annotations
  
  import asyncio
  import json
  from typing import Annotated, Any
  
  from pydantic import Field
  from fastmcp import FastMCP
  
  from http_client import (
      _request_with_retry,
      companies_house_client,
      charity_client,
      gazette_client,

• server.py:164-164 (registration)

  Registration call from server.py — invokes search_fetch.register_tools(mcp) to register the fetch tool on the MCP server.

  search_fetch.register_tools(mcp)