Search UK Due Diligence Registers

Instructions

Implementation Reference

• search_fetch.py:118-145 (handler)

  The 'search' tool handler function. Fans out across Companies House, Charity Commission, disqualified directors, and Gazette insolvency notices in parallel, collects unique prefixed IDs, and returns them.

  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}

• search_fetch.py:106-117 (registration)

  The register_tools function that registers both 'search' and 'fetch' tools with the FastMCP server. Called from server.py 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,
          },
      )

• search_fetch.py:40-47 (helper)

  Private helper that queries Companies House search API and extracts company prefixed IDs.

  async def _company_ids(query: str) -> list[str]:
      async with companies_house_client() as client:
          resp = await _request_with_retry(
              client, "GET", "/search/companies",
              params={"q": query, "items_per_page": 10},
          )
      items = resp.json().get("items") or []
      return [f"company:{item['company_number']}" for item in items if item.get("company_number")]

• search_fetch.py:50-61 (helper)

  Private helper that queries Charity Commission search API and extracts charity prefixed IDs.

  async def _charity_ids(query: str) -> list[str]:
      async with charity_client() as client:
          resp = await _request_with_retry(
              client, "GET", f"/searchCharityName/{query}",
          )
      data = resp.json()
      all_items = data if isinstance(data, list) else []
      return [
          f"charity:{item['reg_charity_number']}"
          for item in all_items[:5]
          if item.get("reg_charity_number") is not None
      ]

• search_fetch.py:64-79 (helper)

  Private helper that queries Companies House disqualified officers search and extracts disqualification prefixed IDs.

  async def _disqualified_ids(query: str) -> list[str]:
      async with companies_house_client() as client:
          resp = await _request_with_retry(
              client, "GET", "/search/disqualified-officers",
              params={"q": query, "items_per_page": 5},
          )
      items = resp.json().get("items") or []
      ids = []
      for item in items:
          links = item.get("links") or {}
          self_link = links.get("self", "") if isinstance(links, dict) else ""
          if self_link:
              officer_id = self_link.rstrip("/").rsplit("/", 1)[-1]
              if officer_id:
                  ids.append(f"disqualification:{officer_id}")
      return ids