> For the complete documentation index, see [llms.txt](https://k-ai.gitbook.io/knowledge-ai/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://k-ai.gitbook.io/knowledge-ai/k-ai-mcp/cookbook.md).

# 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](https://github.com/modelcontextprotocol/python-sdk). Two recipes additionally show TypeScript with `@modelcontextprotocol/sdk`. See the [schema stability policy](/knowledge-ai/k-ai-mcp/schema-stability.md) for our compatibility commitments.

All recipes assume the client has already completed the OAuth flow (see [OAuth flow](/knowledge-ai/k-ai-mcp/oauth-flow.md)) 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.

{% tabs %}
{% tab title="Python" %}

```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"]
```

{% endtab %}

{% tab title="TypeScript" %}

```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 findDocuments(instanceId: string, query: string) {
  const transport = new StreamableHTTPClientTransport(URL, { requestInit: { headers } });
  const client = new Client({ name: "kai-cookbook", version: "1.0.0" });
  await client.connect(transport);

  const hits = await client.callTool({
    name: "retrieval_semantic_nodes_search",
    arguments: { instance_id: instanceId, query },
  });
  const nodes = (hits as any).structuredContent.response;
  const docId = nodes[0].documents[0].id;
  const doc = await client.callTool({
    name: "retrieval_documents_get_document",
    arguments: { instance_id: instanceId, document_id: docId },
  });
  await client.close();
  return (doc as any).structuredContent.response;
}
```

{% endtab %}
{% endtabs %}

## 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.

```python
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.

```python
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.

```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 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"]
```

{% tabs %}
{% tab title="TypeScript" %}

```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;
}
```

{% endtab %}
{% endtabs %}

## 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.

```python
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.

```python
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.