/knowledge-ingest

name: knowledge-ingest
description: "Upload a file (PDF, DOCX, PPTX, XLSX, HTML, EPUB, image) or URL to the Knowledge base. Triggers Marker parsing, chunking, embedding, and async classification. Use when the user says 'index this PDF', 'add this URL to the knowledge base', 'upload these files to Academy', or pastes a file path/URL with ingestion intent."

knowledge-ingest

Group: **Ingestion**. Upload + automatic classification via pipeline parse → chunk → embed → enqueue classify.

When to trigger

• "Index this PDF"
• "Add this URL to the knowledge base"
• "Upload these files to Academy"
• User passes a file path with ingestion intent

Arguments

| Name | Type | Required | Description | |---|---|---|---| | `file_path` | str | one of two | Local path | | `url` | str | one of two | URL to download first | | `connection` | str | no | Defaults to first ready | | `space` | str | yes | Destination space slug | | `unit_id` | str | no | Associated unit | | `title` | str | no | Derived from filename if absent | | `tags` | list[str] | no | User-defined tags |

Workflow

Step 1 — Validate connection + space

If `connection` not provided, use first `ready`. If none: error ("Run `knowledge-admin action=connect`").

Validate space via `GET /spaces`. If not found: list spaces + ask for confirmation.

Step 2 — Resolve file

If `url`:

import requests, tempfile
from pathlib import Path
from urllib.parse import urlparse

parsed = urlparse(url)
filename = Path(parsed.path).name or "downloaded"
tmp = Path(tempfile.gettempdir()) / filename
with requests.get(url, stream=True, timeout=60) as r:
    r.raise_for_status()
    with open(tmp, "wb") as f:
        for chunk in r.iter_content(8192):
            f.write(chunk)
file_path = str(tmp)

If `file_path`: validate existence.

Step 3 — Multipart upload

from dashboard.backend.sdk_client import evo

with open(file_path, "rb") as f:
    result = evo.post(
        "/api/knowledge/v1/documents",
        files={"file": f},
        data={
            "space": space,
            "unit_id": unit_id,
            "title": title or Path(file_path).stem,
            "tags": ",".join(tags or []),
        },
        headers={"X-Knowledge-Connection": connection},
    )

document_id = result["document_id"]

Endpoint returns 202 Accepted + document_id. Async worker.

Step 4 — Poll status

Interval 2s, timeout 10min:

import time
deadline = time.time() + 600
while time.time() < deadline:
    status = evo.get(
        f"/api/knowledge/v1/documents/{document_id}/status",
        headers={"X-Knowledge-Connection": connection},
    )
    phase = status.get("phase")
    if phase in ("done", "ready"):
        break
    if phase == "error":
        raise RuntimeError(status.get("error"))
    time.sleep(2)

Step 5 — Fetch classification (non-blocking)

Classification is asynchronous. 1 extra GET on `/documents/{id}`:

• `content_type != null`: show full classification
• Else: "Classification pending — will appear in seconds via async worker"

Output

✓ Document uploaded: {title}
  document_id: {uuid}
  space: {connection}/{space}
  unit: {unit_title or "none"}
  status: ready
  chunks: {N}
  classification:
    content_type: {lesson|tutorial|faq|...}
    difficulty: {...}
    topics: [...]
  elapsed: {X}s

Actionable failures

• File not found → "File does not exist: `{path}`"
• URL fetch failed → "Download failed: {status_code}"
• Space not found → list available spaces
• Marker models missing → "Run `knowledge-admin action=install-parser`"
• Timeout → "Timeout after 10min. Status: `{phase}`. Check `knowledge-browse`."