Search UK Due Diligence Registers

Instructions

Implementation Reference

• server.py:164-164 (registration)

  Registration of the search tool: calls search_fetch.register_tools(mcp) which registers both 'search' and 'fetch' tools on the MCP server.

  search_fetch.register_tools(mcp)

• search_fetch.py:106-117 (registration)

  Tool registration decorator defining the 'search' tool with metadata annotations.

  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:118-145 (handler)

  Handler function that fans out the query across Companies House, Charity Commission, disqualified directors, and Gazette insolvency notices in parallel, deduplicates results, and returns a list of IDs.

  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:118-120 (schema)

  Input schema: a single 'query' string parameter with length constraints and a description.

  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_fetch.py:40-47 (helper)

  Helper that searches Companies House and returns prefixed IDs (company:{number}).

  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)

  Helper that searches Charity Commission and returns prefixed IDs (charity:{number}).

  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)

  Helper that searches disqualified officers and returns prefixed IDs (disqualification:{officer_id}).

  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

• search_fetch.py:82-99 (helper)

  Helper that searches Gazette insolvency notices and returns prefixed IDs (notice:{notice_id}).

  async def _gazette_ids(query: str) -> list[str]:
      async with gazette_client() as client:
          resp = await _request_with_retry(
              client, "GET", "/insolvency/notice/data.json",
              params={"text": query, "results-page-size": 5},
          )
      raw = resp.json()
      entries = raw.get("entry", []) if isinstance(raw, dict) else []
      if isinstance(entries, dict):
          entries = [entries]
      ids = []
      for entry in entries:
          notice_uri = entry.get("id", "")
          if notice_uri:
              numeric_id = notice_uri.rstrip("/").split("/")[-1]
              if numeric_id:
                  ids.append(f"notice:{numeric_id}")
      return ids