Fetch Full Record from UK Due Diligence Register

Instructions

Implementation Reference

• search_fetch.py:147-227 (handler)

  The `fetch` tool handler function. It takes a prefixed ID (e.g., company:1234), partitions by colon to extract the prefix and value, then routes to the appropriate backend fetch helper (_fetch_company_profile, _fetch_charity_profile, _fetch_disqualified_profile, or _fetch_gazette_notice). Returns a dict with id, title, content (JSON), 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:157-158 (schema)

  Input schema for the `fetch` tool: a single `id` parameter of type string with min_length=3 and max_length=100, described as a prefixed record ID returned by search.

  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)],

• search_fetch.py:106-106 (registration)

  The `register_tools(mcp)` function that registers the `fetch` (and `search`) tool on the FastMCP instance via the `@mcp.tool(name='fetch')` decorator.

  def register_tools(mcp: FastMCP) -> None:

• server.py:164-164 (registration)

  The call to `search_fetch.register_tools(mcp)` in server.py which triggers registration of the `fetch` tool on the main MCP server.

  search_fetch.register_tools(mcp)

• companies_house.py:74-117 (helper)

  Shared helper `_fetch_company_profile` — fetches Companies House company profile and builds a CompanyProfile object (used by the `company` prefix route in the `fetch` tool).

  async def _fetch_company_profile(company_number: str) -> CompanyProfile:
      async with companies_house_client() as client:
          resp = await _request_with_retry(client, "GET", f"/company/{company_number}")
          data = resp.json()
  
          has_charges = False
          try:
              charges_resp = await _request_with_retry(
                  client, "GET", f"/company/{company_number}/charges",
                  params={"items_per_page": 1},
              )
              charges_data = charges_resp.json()
              charges_items = charges_data.get("items") or []
              has_charges = any(
                  item.get("status") == "outstanding" for item in charges_items
              ) or (
                  charges_data.get("total_count", 0) > 0
                  and not charges_items
              )
          except Exception:
              pass
  
      accs_raw = data.get("accounts") or {}
      conf_raw = data.get("confirmation_statement") or {}
  
      return CompanyProfile(
          company_number=str(data.get("company_number") or company_number),
          company_name=data.get("company_name"),
          company_status=data.get("company_status"),
          company_type=data.get("company_type"),
          date_of_creation=data.get("date_of_creation"),
          sic_codes=list(data.get("sic_codes") or []),
          registered_office_address=data.get("registered_office_address") or {},
          has_charges=has_charges,
          accounts=CompanyAccountsSummary(
              overdue=bool(accs_raw.get("overdue", False)),
              last_accounts_made_up_to=(accs_raw.get("last_accounts") or {}).get("made_up_to"),
              next_due=accs_raw.get("next_due"),
          ),
          confirmation_statement=CompanyConfirmationStatementSummary(
              overdue=bool(conf_raw.get("overdue", False)),
              next_due=conf_raw.get("next_due"),
          ),
      )

• charity.py:145-195 (helper)

  Shared helper `_fetch_charity_profile` — fetches Charity Commission details and builds a CharityProfile object (used by the `charity` prefix route).

  async def _fetch_charity_profile(charity_number: str) -> CharityProfile:
      async with charity_client() as client:
          suffix = "0"
          lookup_number = charity_number
          if "-" in charity_number:
              parts = charity_number.split("-", 1)
              lookup_number = parts[0]
              suffix = parts[1]
          resp = await _request_with_retry(
              client, "GET",
              f"/allcharitydetailsV2/{lookup_number}/{suffix}",
          )
          data = resp.json()
  
      reg_num = str(data.get("reg_charity_number") or lookup_number)
      raw_status = data.get("reg_status")
  
      raw_trustees = data.get("trustee_names") or []
      trustees_total = len(raw_trustees)
      trustees = [
          CharityTrustee(trustee_name=t.get("trustee_name"))
          for t in raw_trustees[:30]
          if isinstance(t, dict)
      ]
  
      raw_www = data.get("who_what_where") or []
      www_total = len(raw_www)
      classifications = [
          CharityClassification(
              classification_type=w.get("classification_type"),
              classification_desc=w.get("classification_desc"),
          )
          for w in raw_www[:50]
          if isinstance(w, dict)
      ]
  
      countries_raw = data.get("CharityAoOCountryContinent") or []
      countries = [c.get("country") for c in countries_raw[:10] if isinstance(c, dict) and c.get("country")]
  
      return CharityProfile(
          charity_number=reg_num,
          charity_name=data.get("charity_name"),
          reg_status=raw_status,
          reg_status_label=_STATUS_LABELS.get(raw_status or "", raw_status),
          charity_type=data.get("charity_type"),
          charity_co_reg_number=data.get("charity_co_reg_number") or None,
          date_of_registration=(data.get("date_of_registration") or "")[:10] or None,
          address=_build_address(data),
          latest_income=_coerce_number(data.get("latest_income")),
          latest_expenditure=_coerce_number(data.get("latest_expenditure")),
          insolvent=bool(data.get("insolvent", False)),

• disqualified.py:136-180 (helper)

  Shared helper `_fetch_disqualified_profile` — fetches disqualification records from Companies House (used by the `disqualification` prefix route).

  async def _fetch_disqualified_profile(officer_id: str) -> DisqualifiedProfile:
      oid = officer_id.strip()
      data: dict[str, Any] | None = None
      officer_kind = "natural"
  
      for kind, endpoint in [
          ("natural", f"/disqualified-officers/natural/{oid}"),
          ("corporate", f"/disqualified-officers/corporate/{oid}"),
      ]:
          try:
              async with companies_house_client() as client:
                  resp = await _request_with_retry(client, "GET", endpoint)
                  data = resp.json()
                  officer_kind = kind
                  break
          except httpx.HTTPStatusError as exc:
              if exc.response.status_code == 404:
                  continue
              raise
  
      if data is None:
          raise LookupError(
              f"No disqualification record found for officer ID {oid!r}."
          )
  
      raw_orders = data.get("disqualifications", []) or []
      orders: list[DisqualificationOrder] = []
      for raw in raw_orders:
          company_names = list(raw.get("company_names") or [])
          total_companies = len(company_names)
          truncated = total_companies > 20
          if truncated:
              company_names = company_names[:20]
  
          orders.append(
              DisqualificationOrder(
                  disqualified_from=raw.get("disqualified_from"),
                  disqualified_until=raw.get("disqualified_until"),
                  reason=raw.get("reason") or {},
                  company_names=company_names,
                  company_names_truncated=truncated,
                  company_names_total=total_companies,
                  address=raw.get("address") or {},
                  case_identifier=raw.get("case_identifier"),
                  heard_on=raw.get("heard_on"),

• gazette.py:138-143 (helper)

  Shared helper `_fetch_gazette_notice` — fetches Gazette notice JSON-LD data (used by the `notice` prefix route).

  async def _fetch_gazette_notice(notice_id: str) -> dict:
      url = f"https://www.thegazette.co.uk/notice/{notice_id.strip()}/data.json?view=linked-data"
      async with httpx.AsyncClient(timeout=15.0) as client:
          resp = await client.get(url, headers={"Accept": "application/json"})
          resp.raise_for_status()
          return resp.json()