get_corpus_status
Check the current status of documentary corpora in KFabric to monitor data quality and readiness for RAG implementation.
Instructions
Read the latest corpus status for a query or corpus id.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| kwargs | Yes |
Implementation Reference
- kfabric/mcp/registry.py:183-198 (handler)The implementation of the get_corpus_status tool handler.
def _get_corpus_status(_orchestrator: Orchestrator, db: Session, _settings: AppSettings, arguments: dict[str, Any]) -> Any: corpus_id = arguments.get("corpus_id") if corpus_id: corpus = db.get(Corpus, corpus_id) else: query_id = arguments["query_id"] corpus = db.scalars(select(Corpus).where(Corpus.query_id == query_id).order_by(Corpus.created_at.desc())).first() if not corpus: raise ValueError("Corpus not found") return { "corpus_id": corpus.id, "status": corpus.status, "index_ready": corpus.index_ready, "artifact_path": corpus.artifact_path, } - kfabric/mcp/registry.py:280-295 (registration)Registration of the get_corpus_status tool with its schema definition and handler.
ToolDefinition( name="get_corpus_status", title="Get Corpus Status", description="Read the latest corpus status for a query or corpus id.", version="1.0.0", input_schema={ "type": "object", "properties": { "query_id": {"type": "string"}, "corpus_id": {"type": "string"}, }, }, output_schema={"type": "object"}, security=common_security, handler=_get_corpus_status, ),