Search GOV.UK

Instructions

Implementation Reference

• govuk_mcp/server.py:180-273 (registration)

  Tool 'govuk_search' is registered via @mcp.tool decorator on line 184, which sets the name='govuk_search' and annotations including readOnlyHint and idempotentHint.

  # ---------------------------------------------------------------------------
  # Tools
  # ---------------------------------------------------------------------------
  
  @mcp.tool(
      name="govuk_search",
      annotations={
          "title": "Search GOV.UK",
          "readOnlyHint": True,
          "destructiveHint": False,
          "idempotentHint": True,
          "openWorldHint": True,
      },
  )
  @_timed_tool
  async def govuk_search(
      query: Annotated[str, Field(description="Free-text search query, e.g. 'universal credit eligibility' or 'MOT check'", min_length=1, max_length=500)],
      ctx: Context,
      count: Annotated[int, Field(description="Number of results to return (1–50)", ge=1, le=MAX_COUNT)] = 10,
      start: Annotated[int, Field(description="Offset for pagination, e.g. 10 for the second page of 10 results", ge=0)] = 0,
      filter_format: Annotated[Optional[str], Field(description="Filter by document format. Common values: 'guide', 'answer', 'transaction', 'publication', 'news_article', 'detailed_guide', 'hmrc_manual_section', 'travel_advice', 'organisation'. Leave blank to search all types.")] = None,
      filter_organisations: Annotated[Optional[str], Field(description="Filter by organisation slug, e.g. 'hm-revenue-customs', 'department-for-work-pensions', 'driver-and-vehicle-standards-agency'.")] = None,
      order: Annotated[Optional[str], Field(description="Sort order. Use '-public_timestamp' for newest-first (default relevance).")] = None,
  ) -> GovukSearchResult:
      """Search GOV.UK's 700k+ content items using the official Search API.
  
      Returns a list of matching content items with title, description, link,
      format, owning organisation(s), and last updated timestamp.
  
      Use filter_format to narrow to specific content types (e.g. 'transaction'
      for citizen-facing services, 'guide' for guidance, 'publication' for
      official documents). Use filter_organisations to restrict to a department.
      """
      client = _client(ctx)
  
      query_params: dict[str, Any] = {
          "q": query,
          "count": count,
          "start": start,
          "fields[]": ["title", "description", "link", "format", "organisations", "public_timestamp"],
      }
      if filter_format:
          query_params["filter_document_type"] = filter_format
      if filter_organisations:
          query_params["filter_organisations"] = filter_organisations
      if order:
          query_params["order"] = order
  
      resp = await client.get(SEARCH_BASE, params=query_params)
      resp.raise_for_status()
      data = resp.json()
  
      raw_results = data.get("results", [])
      results: list[GovukSearchResultItem] = []
      for r in raw_results:
          orgs = [
              GovukSearchOrganisation(
                  title=o.get("title"),
                  acronym=o.get("acronym"),
                  slug=o.get("slug"),
              )
              for o in r.get("organisations", [])
          ]
          link = r.get("link") or ""
          base = link.lstrip("/")
          results.append(
              GovukSearchResultItem(
                  title=r.get("title"),
                  description=r.get("description"),
                  link=link or None,
                  url=f"https://www.gov.uk{link}" if link else None,
                  format=r.get("format"),
                  organisations=orgs,
                  public_timestamp=r.get("public_timestamp"),
                  next_steps=({
                      "get_content": f"govuk_get_content(base_path={base!r})",
                      "grep": f"govuk_grep_content(base_path={base!r}, pattern=...)",
                  } if link else {}),
              )
          )
  
      total = data.get("total", 0) or 0
      returned = len(results)
      has_more = (start + returned) < total if isinstance(total, int) else returned == count
  
      return GovukSearchResult(
          query=query,
          total=total,
          start=start,
          count=count,
          returned=returned,
          has_more=has_more,
          results=results,
      )

• govuk_mcp/server.py:195-273 (handler)

  The async function govuk_search() is the handler. It accepts query, count, start, filter_format, filter_organisations, order params, calls the GOV.UK Search API at https://www.gov.uk/api/search.json, parses results, and returns a GovukSearchResult model.

  async def govuk_search(
      query: Annotated[str, Field(description="Free-text search query, e.g. 'universal credit eligibility' or 'MOT check'", min_length=1, max_length=500)],
      ctx: Context,
      count: Annotated[int, Field(description="Number of results to return (1–50)", ge=1, le=MAX_COUNT)] = 10,
      start: Annotated[int, Field(description="Offset for pagination, e.g. 10 for the second page of 10 results", ge=0)] = 0,
      filter_format: Annotated[Optional[str], Field(description="Filter by document format. Common values: 'guide', 'answer', 'transaction', 'publication', 'news_article', 'detailed_guide', 'hmrc_manual_section', 'travel_advice', 'organisation'. Leave blank to search all types.")] = None,
      filter_organisations: Annotated[Optional[str], Field(description="Filter by organisation slug, e.g. 'hm-revenue-customs', 'department-for-work-pensions', 'driver-and-vehicle-standards-agency'.")] = None,
      order: Annotated[Optional[str], Field(description="Sort order. Use '-public_timestamp' for newest-first (default relevance).")] = None,
  ) -> GovukSearchResult:
      """Search GOV.UK's 700k+ content items using the official Search API.
  
      Returns a list of matching content items with title, description, link,
      format, owning organisation(s), and last updated timestamp.
  
      Use filter_format to narrow to specific content types (e.g. 'transaction'
      for citizen-facing services, 'guide' for guidance, 'publication' for
      official documents). Use filter_organisations to restrict to a department.
      """
      client = _client(ctx)
  
      query_params: dict[str, Any] = {
          "q": query,
          "count": count,
          "start": start,
          "fields[]": ["title", "description", "link", "format", "organisations", "public_timestamp"],
      }
      if filter_format:
          query_params["filter_document_type"] = filter_format
      if filter_organisations:
          query_params["filter_organisations"] = filter_organisations
      if order:
          query_params["order"] = order
  
      resp = await client.get(SEARCH_BASE, params=query_params)
      resp.raise_for_status()
      data = resp.json()
  
      raw_results = data.get("results", [])
      results: list[GovukSearchResultItem] = []
      for r in raw_results:
          orgs = [
              GovukSearchOrganisation(
                  title=o.get("title"),
                  acronym=o.get("acronym"),
                  slug=o.get("slug"),
              )
              for o in r.get("organisations", [])
          ]
          link = r.get("link") or ""
          base = link.lstrip("/")
          results.append(
              GovukSearchResultItem(
                  title=r.get("title"),
                  description=r.get("description"),
                  link=link or None,
                  url=f"https://www.gov.uk{link}" if link else None,
                  format=r.get("format"),
                  organisations=orgs,
                  public_timestamp=r.get("public_timestamp"),
                  next_steps=({
                      "get_content": f"govuk_get_content(base_path={base!r})",
                      "grep": f"govuk_grep_content(base_path={base!r}, pattern=...)",
                  } if link else {}),
              )
          )
  
      total = data.get("total", 0) or 0
      returned = len(results)
      has_more = (start + returned) < total if isinstance(total, int) else returned == count
  
      return GovukSearchResult(
          query=query,
          total=total,
          start=start,
          count=count,
          returned=returned,
          has_more=has_more,
          results=results,
      )

• govuk_mcp/models.py:16-89 (schema)

  Schema definitions for govuk_search: GovukSearchOrganisation (lines 20-27), GovukSearchResultItem (lines 30-63), and GovukSearchResult (lines 66-89) output models with Pydantic Field descriptions.

  # govuk_search — Shape B (paginated search)
  # ---------------------------------------------------------------------------
  
  
  class GovukSearchOrganisation(BaseModel):
      """Organisation owning a search result."""
  
      model_config = ConfigDict(str_strip_whitespace=True)
  
      title: Optional[str] = Field(None, description="Full organisation title, e.g. 'HM Revenue & Customs'.")
      acronym: Optional[str] = Field(None, description="Organisation acronym, e.g. 'HMRC'.")
      slug: Optional[str] = Field(None, description="Organisation slug for use with govuk_get_organisation.")
  
  
  class GovukSearchResultItem(BaseModel):
      """A single GOV.UK search hit."""
  
      model_config = ConfigDict(str_strip_whitespace=True)
  
      title: Optional[str] = Field(None, description="Page title.")
      description: Optional[str] = Field(None, description="Short human-readable summary of the page.")
      link: Optional[str] = Field(
          None,
          description="GOV.UK relative path for the page, e.g. '/universal-credit'. Use as base_path in govuk:// resource URIs — the next_steps field constructs them for you.",
      )
      url: Optional[str] = Field(None, description="Absolute https://www.gov.uk URL for the page.")
      format: Optional[str] = Field(
          None,
          description="Document format, e.g. 'guide', 'answer', 'transaction', 'publication', 'news_article'.",
      )
      organisations: list[GovukSearchOrganisation] = Field(
          default_factory=list,
          description="Owning organisation(s) for the page.",
      )
      public_timestamp: Optional[str] = Field(
          None,
          description="ISO-8601 timestamp for when this page was last publicly updated.",
      )
      next_steps: dict[str, str] = Field(
          default_factory=dict,
          description=(
              "Canonical resource URIs and tool for reading this content item. "
              "Read `header` first for orientation, then `index` to discover sections, "
              "then `section_template` (substitute the anchor from the index) for "
              "specific sections. Use `grep_tool` for content discovery instead of "
              "reading every section."
          ),
      )
  
  
  class GovukSearchResult(BaseModel):
      """A page of GOV.UK search results."""
  
      model_config = ConfigDict(str_strip_whitespace=True)
  
      query: str = Field(..., description="The free-text query that was searched.")
      total: int = Field(..., description="Total matching results across all pages on GOV.UK.")
      start: int = Field(..., description="Offset used for this page (zero-based).")
      count: int = Field(..., description="Max results requested for this page.")
      returned: int = Field(..., description="Number of results actually returned in this response.")
      has_more: bool = Field(
          ...,
          description=(
              "True if more results exist beyond this page. Re-call with "
              "start=start+returned to fetch the next page."
          ),
      )
      results: list[GovukSearchResultItem] = Field(
          default_factory=list,
          description=(
              "Matching pages. Use the `link` field of any result as the "
              "`base_path` input to govuk_get_content for the full item."
          ),
      )

• govuk_mcp/server.py:40-40 (helper)

  Constant SEARCH_BASE = 'https://www.gov.uk/api/search.json' used by the handler.

  SEARCH_BASE = "https://www.gov.uk/api/search.json"

• govuk_mcp/server.py:64-82 (helper)

  The _timed_tool decorator wraps the handler to capture Prometheus metrics (call count and duration).

  def _timed_tool(fn):
      tool_name = fn.__name__
  
      @functools.wraps(fn)
      async def wrapped(*args, **kwargs):
          t0 = time.perf_counter()
          try:
              result = await fn(*args, **kwargs)
              tool_calls_total.labels(tool_name, TRANSPORT, REGION, "ok").inc()
              return result
          except BaseException:
              tool_calls_total.labels(tool_name, TRANSPORT, REGION, "error").inc()
              raise
          finally:
              tool_duration_seconds.labels(tool_name, TRANSPORT, REGION).observe(
                  time.perf_counter() - t0
              )
  
      return wrapped