Cookbook

Multi-tool recipes combining the K-AI Retrieval and Audit MCP servers. Each recipe shows the end-to-end MCP tool sequence with sample Python calls using the mcp Python SDK. Two recipes additionally show TypeScript with @modelcontextprotocol/sdk. See the schema stability policy for our compatibility commitments.

All recipes assume the client has already completed the OAuth flow (see OAuth flow) and stored an access token. In the snippets below, ACCESS_TOKEN is the bearer token; the SDK transport handles refresh.

K-AI MCP tools wrap every response as {"response": <payload>}. With the Python mcp SDK, the typed payload is exposed on result.structuredContent; the snippets below access it as result.structuredContent["response"]. With @modelcontextprotocol/sdk for TypeScript the typed payload also lives on result.structuredContent (cast as any to read it).

Recipe 1 — Find documents on a topic

Goal. A host LLM helps a Consumer find documents in a K-AI instance relevant to a topic.

Tools used (Retrieval, stable).

1. retrieval_semantic_nodes_search — semantic search over the Neural Semantic Graph.

2. retrieval_documents_get_document — fetch the full content of a matching document.

Python

import os
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

URL = "https://api-retrieval.kai-studio.ai/mcp"
HEADERS = {"Authorization": f"Bearer {os.environ['ACCESS_TOKEN']}"}

async def find_documents(instance_id: str, query: str):
    async with streamablehttp_client(URL, headers=HEADERS) as (r, w, _):
        async with ClientSession(r, w) as session:
            await session.initialize()
            hits = await session.call_tool(
                "retrieval_semantic_nodes_search",
                {"instance_id": instance_id, "query": query},
            )
            nodes = hits.structuredContent["response"]
            # Each node carries a `documents` array; pick the first document of the
            # top-ranked node.
            doc_id = nodes[0]["documents"][0]["id"]
            doc = await session.call_tool(
                "retrieval_documents_get_document",
                {"instance_id": instance_id, "document_id": doc_id},
            )
            return doc.structuredContent["response"]

TypeScript

Recipe 2 — Resolve an open audit conflict from an LLM

Goal. An LLM agent helps a Document Steward resolve a conflict surfaced by K-AI Audit: orient, pick a conflict, record the expert answer, then close it once the Steward confirms the recommended document edits were applied.

Tools used (Audit).

1. audit_dashboard_get — orient on all instances (open conflicts, pending questions, missing subjects).

2. audit_full_audit_conflict_list — list open full-audit conflicts (defaults to DETECTED).

3. audit_full_audit_conflict_get — inspect a conflict (sources, evidence).

4. audit_full_audit_conflict_set_answer — record the expert's resolved answer; the response carries per-document modification recommendations.

5. audit_full_audit_conflict_close — close the conflict only after the Steward confirms the recommended modifications were applied to the source documents.

import os
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

URL = "https://api-audit.kai-studio.ai/mcp"
HEADERS = {"Authorization": f"Bearer {os.environ['ACCESS_TOKEN']}"}

async def resolve_first_open_conflict(instance_id: str, expert_answer: str):
    async with streamablehttp_client(URL, headers=HEADERS) as (r, w, _):
        async with ClientSession(r, w) as session:
            await session.initialize()

            await session.call_tool("audit_dashboard_get", {})

            conflicts = await session.call_tool(
                "audit_full_audit_conflict_list",
                {"instance_id": instance_id, "limit": 1},
            )
            conflict_id = conflicts.structuredContent["response"][0]["id"]

            await session.call_tool(
                "audit_full_audit_conflict_get",
                {"instance_id": instance_id, "conflict_id": conflict_id},
            )

            recos = await session.call_tool(
                "audit_full_audit_conflict_set_answer",
                {
                    "instance_id": instance_id,
                    "conflict_id": conflict_id,
                    "answer": expert_answer,
                },
            )
            recommendations = recos.structuredContent["response"]["recommendations"]
            # Render `recommendations` as a markdown table with document_url links.
            # Ask the expert to confirm modifications applied, then:
            await session.call_tool(
                "audit_full_audit_conflict_close",
                {"instance_id": instance_id, "conflict_id": conflict_id},
            )

Recipe 3 — Multi-instance search across an org

Goal. Search across every K-AI instance the user has access to in one go, then aggregate.

Tools used (Retrieval, stable).

1. retrieval_instances_list_available_instances — list instances visible to the user.

2. retrieval_semantic_nodes_search — call once per instance.

import asyncio, os
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

URL = "https://api-retrieval.kai-studio.ai/mcp"
HEADERS = {"Authorization": f"Bearer {os.environ['ACCESS_TOKEN']}"}

async def search_org(query: str):
    async with streamablehttp_client(URL, headers=HEADERS) as (r, w, _):
        async with ClientSession(r, w) as session:
            await session.initialize()
            listing = await session.call_tool(
                "retrieval_instances_list_available_instances", {},
            )
            instances = listing.structuredContent["response"]["instances"]

            async def one(inst):
                hits = await session.call_tool(
                    "retrieval_semantic_nodes_search",
                    {"instance_id": inst["id"], "query": query},
                )
                return hits.structuredContent["response"]

            results = await asyncio.gather(*(one(i) for i in instances))
            return [(inst["id"], nodes) for inst, nodes in zip(instances, results)]

Recipe 4 — Fetch documents by ID across instances

Goal. A host LLM has accumulated document IDs (from a prior search or a user-provided list) and needs the full content in a single round-trip per instance.

Tools used (Retrieval, stable).

1. retrieval_documents_list_by_ids — batch fetch.

import os
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

URL = "https://api-retrieval.kai-studio.ai/mcp"
HEADERS = {"Authorization": f"Bearer {os.environ['ACCESS_TOKEN']}"}

async def fetch_batch(instance_id: str, document_ids: list[str]):
    async with streamablehttp_client(URL, headers=HEADERS) as (r, w, _):
        async with ClientSession(r, w) as session:
            await session.initialize()
            result = await session.call_tool(
                "retrieval_documents_list_by_ids",
                {"instance_id": instance_id, "document_ids": document_ids},
            )
            return result.structuredContent["response"]

TypeScript

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";

const URL = new URL("https://api-retrieval.kai-studio.ai/mcp");
const headers = { Authorization: `Bearer ${process.env.ACCESS_TOKEN}` };

export async function fetchBatch(instanceId: string, documentIds: string[]) {
  const transport = new StreamableHTTPClientTransport(URL, { requestInit: { headers } });
  const client = new Client({ name: "kai-cookbook", version: "1.0.0" });
  await client.connect(transport);

  const docs = await client.callTool({
    name: "retrieval_documents_list_by_ids",
    arguments: { instance_id: instanceId, document_ids: documentIds },
  });
  await client.close();
  return (docs as any).structuredContent.response;
}

Recipe 5 — Surface missing subjects to a Steward

Goal. Help a Document Steward review documentary gaps (topics not covered by the Repository) on a K-AI Audit instance and the questions that triggered the gap detection.

Tools used (Audit).

1. audit_missing_subject_list — list known missing subjects for an instance.

2. audit_missing_subject_find — re-run the missing-subject detection.

3. audit_missing_subject_list_questions — list the questions that surfaced a given missing subject.

import os
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

URL = "https://api-audit.kai-studio.ai/mcp"
HEADERS = {"Authorization": f"Bearer {os.environ['ACCESS_TOKEN']}"}

async def review_missing_subjects(instance_id: str):
    async with streamablehttp_client(URL, headers=HEADERS) as (r, w, _):
        async with ClientSession(r, w) as session:
            await session.initialize()

            # Refresh detection (optional — may be expensive).
            await session.call_tool(
                "audit_missing_subject_find", {"instance_id": instance_id},
            )

            listing = await session.call_tool(
                "audit_missing_subject_list",
                {"instance_id": instance_id, "offset": 0, "limit": 50},
            )
            for subj in listing.structuredContent["response"]:
                questions = await session.call_tool(
                    "audit_missing_subject_list_questions",
                    {"instance_id": instance_id, "subject_id": subj["id"]},
                )
                yield subj, questions.structuredContent["response"]

Recipe 6 — Cross-check a draft answer against the audited base (Retrieval + Audit)

Goal. A host LLM has drafted an answer from Retrieval results. Before returning it, validate that no unresolved K-AI Audit conflict on the same topic would contradict it.

Tools used.

1. retrieval_semantic_nodes_search — find the supporting passages for the draft.

2. audit_full_audit_conflict_list — list open full-audit conflicts in the same instance (defaults to DETECTED). Filter by topic client-side on the returned subject / explanation fields.

import os
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

RET = "https://api-retrieval.kai-studio.ai/mcp"
AUD = "https://api-audit.kai-studio.ai/mcp"
HEADERS = {"Authorization": f"Bearer {os.environ['ACCESS_TOKEN']}"}

async def cross_check(instance_id: str, topic: str):
    async with streamablehttp_client(RET, headers=HEADERS) as (r, w, _):
        async with ClientSession(r, w) as ret:
            await ret.initialize()
            passages_result = await ret.call_tool(
                "retrieval_semantic_nodes_search",
                {"instance_id": instance_id, "query": topic},
            )
            passages = passages_result.structuredContent["response"]

    async with streamablehttp_client(AUD, headers=HEADERS) as (r, w, _):
        async with ClientSession(r, w) as aud:
            await aud.initialize()
            conflicts_result = await aud.call_tool(
                "audit_full_audit_conflict_list",
                {"instance_id": instance_id},
            )
            all_open = conflicts_result.structuredContent["response"]
            # Client-side keyword filter on the returned conflicts.
            needle = topic.lower()
            open_conflicts = [
                c for c in all_open
                if needle in (c.get("subject") or "").lower()
                or needle in (c.get("explanation") or "").lower()
            ]

    return {"passages": passages, "open_conflicts": open_conflicts}

If open_conflicts is non-empty, the host LLM should warn the Consumer that the topic has unresolved contradictions in the audited base before presenting the drafted answer.