Search Committee Evidence

Instructions

Implementation Reference

• src/modules/committees/tools.py:206-304 (handler)

  The async function committees_search_evidence — the actual handler. Uses CommitteeEvidenceInput schema, fetches from committees-api.parliament.uk (OralEvidence and WrittenEvidence endpoints), caps titles per max_title_chars, and returns a CommitteeEvidencePage with has_more pagination flag.

  async def committees_search_evidence(params: CommitteeEvidenceInput, ctx: Context) -> CommitteeEvidencePage:
      """Search oral and written evidence submitted to a parliamentary committee.
  
      Returns ONE PAGE of evidence (default 20). Free-text titles are capped
      per max_title_chars; witness lists are capped at 10 per item. For
      committees with many submissions, re-call with offset=offset+returned
      while has_more is true.
  
      Args:
          params: CommitteeEvidenceInput with committee_id, evidence_type,
              offset/limit pagination, and max_title_chars.
      """
      client: httpx.AsyncClient = ctx.lifespan_context["http"]
  
      def _cap_title(t: str) -> str:
          if len(t) > params.max_title_chars:
              return t[: params.max_title_chars] + " …[truncated]"
          return t
  
      async def fetch_oral(skip: int, take: int) -> tuple[list[EvidenceItem], int]:
          resp = await client.get(
              f"{COMMITTEES_BASE}/OralEvidence",
              params={"CommitteeId": params.committee_id, "Skip": skip, "Take": take},
          )
          resp.raise_for_status()
          data = resp.json()
          items = data.get("items", data.get("results", data)) if isinstance(data, dict) else data
          if not isinstance(items, list):
              return [], 0
          results: list[EvidenceItem] = []
          for item in items:
              ev_date = item.get("evidenceDate") or item.get("date")
              witnesses: list[str] = []
              for w in item.get("witnesses", []):
                  if isinstance(w, str):
                      witnesses.append(w)
                  elif isinstance(w, dict):
                      witnesses.append(w.get("name", str(w)))
              results.append(EvidenceItem(
                  id=item.get("id", 0),
                  type="oral",
                  title=_cap_title(item.get("title", item.get("sessionTitle", "Oral evidence session"))),
                  date=date.fromisoformat(ev_date[:10]) if ev_date else None,
                  witnesses=(witnesses[:10] or None),
                  url=item.get("url"),
              ))
          return results, len(items)
  
      async def fetch_written(skip: int, take: int) -> tuple[list[EvidenceItem], int]:
          resp = await client.get(
              f"{COMMITTEES_BASE}/WrittenEvidence",
              params={"CommitteeId": params.committee_id, "Skip": skip, "Take": take},
          )
          resp.raise_for_status()
          data = resp.json()
          items = data.get("items", data.get("results", data)) if isinstance(data, dict) else data
          if not isinstance(items, list):
              return [], 0
          results: list[EvidenceItem] = []
          for item in items:
              ev_date = item.get("dateReceived") or item.get("date")
              results.append(EvidenceItem(
                  id=item.get("id", 0),
                  type="written",
                  title=_cap_title(item.get("title", "Written evidence")),
                  date=date.fromisoformat(ev_date[:10]) if ev_date else None,
                  witnesses=None,
                  url=item.get("url"),
              ))
          return results, len(items)
  
      evidence: list[EvidenceItem] = []
      has_more = False
  
      if params.evidence_type == "oral":
          evidence, raw = await fetch_oral(params.offset, params.limit)
          has_more = raw == params.limit
      elif params.evidence_type == "written":
          evidence, raw = await fetch_written(params.offset, params.limit)
          has_more = raw == params.limit
      else:
          oral_take = (params.limit + 1) // 2  # remainder to oral
          written_take = params.limit // 2
          (oral, oral_raw), (written, written_raw) = await asyncio.gather(
              fetch_oral(params.offset, oral_take),
              fetch_written(params.offset, written_take),
          )
          evidence = oral + written
          has_more = (oral_raw == oral_take) or (written_raw == written_take)
  
      return CommitteeEvidencePage(
          committee_id=params.committee_id,
          evidence_type=params.evidence_type,
          offset=params.offset,
          limit=params.limit,
          returned=len(evidence),
          has_more=has_more,
          evidence=evidence,
      )

• src/modules/committees/tools.py:50-82 (schema)

  CommitteeEvidenceInput Pydantic model — input schema for the tool. Fields: committee_id, evidence_type (oral/written/both), offset, limit, max_title_chars.

  class CommitteeEvidenceInput(BaseModel):
      model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")
  
      committee_id: int = Field(..., description="Committee ID from committees_search_committees results.", ge=1)
      evidence_type: Literal["oral", "written", "both"] = Field("both", description="Type of evidence to search.")
      offset: int = Field(
          0,
          ge=0,
          le=2000,
          description=(
              "Number of evidence items to skip before this page. Default 0. "
              "Re-call with offset=offset+returned while has_more is true."
          ),
      )
      limit: int = Field(
          20,
          ge=1,
          le=100,
          description=(
              "Maximum evidence items to return. Default 20. When evidence_type='both' "
              "the limit is split across oral and written (roughly half each)."
          ),
      )
      max_title_chars: int = Field(
          300,
          ge=50,
          le=2000,
          description=(
              "Per-item cap on the free-text title field. Default 300 prevents "
              "context blow-up from verbose inquiry titles. Raise to 1000+ only "
              "when you need the full title text."
          ),
      )

• src/modules/committees/models.py:54-99 (schema)

  EvidenceItem and CommitteeEvidencePage Pydantic models — output schema. EvidenceItem has id, type, title, date, witnesses, url. CommitteeEvidencePage has committee_id, evidence_type, offset, limit, returned, has_more, evidence.

  class EvidenceItem(BaseModel):
      model_config = ConfigDict(str_strip_whitespace=True)
  
      id: int = Field(..., description="Evidence item ID")
      type: Literal["oral", "written"] = Field(..., description="Type of evidence")
      title: str = Field(..., description="Evidence title or session description (may be truncated per max_title_chars)")
      date: Date | None = Field(None, description="Date the evidence was given or submitted")
      witnesses: list[str] | None = Field(None, description="Witness names (oral evidence only, capped at 10 per item)")
      url: str | None = Field(None, description="URL to the evidence document")
  
  
  class CommitteeEvidencePage(BaseModel):
      """A page of evidence submissions to a parliamentary committee.
  
      Returned by committees_search_evidence. Callers paginate by
      re-calling with offset=offset+returned while has_more is True.
      When evidence_type="both", oral and written evidence are
      interleaved in a single `evidence` list and the limit is split
      across both.
      """
  
      model_config = ConfigDict(str_strip_whitespace=True)
  
      committee_id: int = Field(..., description="Committee ID this page belongs to")
      evidence_type: Literal["oral", "written", "both"] = Field(
          ..., description="Evidence type filter applied to this query"
      )
      offset: int = Field(..., description="Number of evidence items skipped before this page")
      limit: int = Field(..., description="Max evidence items requested for this page")
      returned: int = Field(..., description="Number of evidence items actually returned in this call")
      has_more: bool = Field(
          ...,
          description=(
              "True if there may be more evidence beyond this page. Re-call with "
              "offset=offset+returned to fetch the next page. Conservative: when "
              "evidence_type='both', True if either oral or written upstream page "
              "came back full."
          ),
      )
      evidence: list[EvidenceItem] = Field(
          default_factory=list,
          description=(
              "Evidence items in this page. Titles are capped per max_title_chars; "
              "witness lists are capped at 10 per item."
          ),
      )

• src/modules/committees/tools.py:202-205 (registration)

  Registration decorator: @mcp.tool(name='search_evidence', ...) which registers the tool with FastMCP under the name 'search_evidence' (mounted under 'committees' namespace, so full name is 'committees_search_evidence').

  @mcp.tool(
      name="search_evidence",
      annotations={"title": "Search Committee Evidence", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True},
  )

• src/modules/committees/__init__.py:8-23 (registration)

  Module-level FastMCP instance creation and tool registration. committees_mcp is created, middleware added, and register_tools(committees_mcp) is called to register all tools. This MCP is then mounted at gateway.py line 189 under namespace 'committees'.

  committees_mcp = FastMCP(
      name="committees",
      instructions=(
          "Search UK parliamentary select committees and their evidence submissions. "
          "Use committees_search_committees to find committees by name, house, or status. "
          "Use committees_get_committee to get detail including current membership. "
          "Use committees_search_evidence to find oral and written evidence for a committee. "
          "All data from committees-api.parliament.uk. No authentication required."
      ),
  )
  
  committees_mcp.add_middleware(ResponseCachingMiddleware(call_tool_settings=CallToolSettings(ttl=3600)))
  
  register_tools(committees_mcp)
  
  __all__ = ["committees_mcp"]