KFabric

Instructions

Implementation Reference

• kfabric/mcp/registry.py:132-143 (handler)

  The MCP handler that processes the 'list_candidates' tool request by calling the orchestrator's list_candidates method.

  def _list_candidates(orchestrator: Orchestrator, _db: Session, _settings: AppSettings, arguments: dict[str, Any]) -> Any:
      query_id = arguments["query_id"]
      return [
          {
              "id": candidate.id,
              "title": candidate.title,
              "status": candidate.status,
              "domain": candidate.domain,
              "source_url": candidate.source_url,
          }
          for candidate in orchestrator.list_candidates(query_id)
      ]

• kfabric/mcp/registry.py:214-222 (registration)

  Registration of the 'list_candidates' tool in the MCP registry.

      name="list_candidates",
      title="List Candidates",
      description="List candidate documents for a query.",
      version="1.0.0",
      input_schema={"type": "object", "properties": {"query_id": {"type": "string"}}, "required": ["query_id"]},
      output_schema={"type": "array"},
      security=common_security,
      handler=_list_candidates,
  ),

• kfabric/services/orchestrator.py:111-112 (handler)

  The business logic implementation of list_candidates within the Orchestrator service.

  def list_candidates(self, query_id: str) -> list[CandidateDocument]:
      return list(self.session.scalars(select(CandidateDocument).where(CandidateDocument.query_id == query_id).order_by(CandidateDocument.discovery_rank)))