diff --git a/README.md b/README.md index 502c0e4a..c12b3ee7 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,7 @@ Open-source toolkit for **Pipefy** developers: a Model Context Protocol (MCP) se | Component | Package / path | Purpose | |-----------|----------------|---------| -| **MCP server** | `pipefy-mcp-server` | Exposes **168** tools to MCP clients (Cursor, Claude Desktop, Claude Code, and others). | +| **MCP server** | `pipefy-mcp-server` | Exposes **172** tools to MCP clients (Cursor, Claude Desktop, Claude Code, and others). | | **CLI** | `pipefy-cli` | Terminal commands aligned with MCP capabilities; see [`docs/parity.md`](docs/parity.md). | | **SDK** | `pipefy` | Vendor GraphQL client, services, and models shared by MCP and CLI. | | **Skills** | [`skills/`](skills/) | Markdown playbooks (Anthropic Skills format) for common Pipefy workflows. | @@ -141,7 +141,7 @@ Deprecation and semver (post-1.0): [`docs/DEPRECATION.md`](docs/DEPRECATION.md). ## MCP server -The server registers **168 tools** across thirteen domains. Canonical names: `PIPEFY_TOOL_NAMES` in [`packages/mcp/src/pipefy_mcp/tools/registry.py`](packages/mcp/src/pipefy_mcp/tools/registry.py). +The server registers **172 tools** across thirteen domains. Canonical names: `PIPEFY_TOOL_NAMES` in [`packages/mcp/src/pipefy_mcp/tools/registry.py`](packages/mcp/src/pipefy_mcp/tools/registry.py). Tool descriptions and `Args:` blocks come from Python docstrings (what MCP clients show to models). Per-area reference docs cover parameters, edge cases, and cross-cutting behavior. @@ -155,7 +155,7 @@ Tool descriptions and `Args:` blocks come from Python docstrings (what MCP clien | **Reports** | 17 | Pipe and organization reports, async exports. | [docs](docs/mcp/tools/reports.md) | | **Automations & AI** | 23 | Automations, AI automations, AI agents, validators. | [docs](docs/mcp/tools/automations-and-ai.md) | | **LLM providers** | 5 | Discovery reads for the organization's LLM providers (custom + Pipefy-managed), vendor model lists, owner defaults, dependencies, and a read-access probe. | [docs](docs/mcp/tools/llm-providers.md) | -| **Knowledge bases** | 6 | Pipe-scoped AI knowledge bases: list all items, plain text CRUD, and a read-access probe. Attach sources to agents/behaviors via `dataSourceIds`. | [docs](docs/mcp/tools/knowledge-bases.md) | +| **Knowledge bases** | 10 | Pipe-scoped AI knowledge bases: list all items, plain text and document CRUD (one-shot PDF upload), and a read-access probe. Attach sources to agents/behaviors via `dataSourceIds`. | [docs](docs/mcp/tools/knowledge-bases.md) | | **iPaaS** | 4 | Lazy discovery, invocation, and app-connection setup for a pipe's iPaaS (Advanced Automations) workspace (`get_ipaas_tools`, `call_ipaas_tool`, plus the connection meta-tools). | [docs](docs/mcp/tools/ipaas.md) | | **Observability** | 11 | Logs, usage, credits, execution metrics, job exports. | [docs](docs/mcp/tools/observability.md) | | **Members, email & webhooks** | 11 | Membership, inbox email, webhooks. | [docs](docs/mcp/tools/members-email-webhooks.md) | diff --git a/docs/mcp/README.md b/docs/mcp/README.md index 55b7f80f..99c01c7b 100644 --- a/docs/mcp/README.md +++ b/docs/mcp/README.md @@ -13,7 +13,7 @@ Material in this tree describes **`pipefy-mcp-server`**: the MCP process, tool b | [`tools/reports.md`](tools/reports.md) | Pipe and organization reports, async exports | | [`tools/automations-and-ai.md`](tools/automations-and-ai.md) | Traditional automations, AI automations, AI agents, validators | | [`tools/llm-providers.md`](tools/llm-providers.md) | LLM provider discovery: custom + system providers, vendor models, defaults, dependencies, access probe | -| [`tools/knowledge-bases.md`](tools/knowledge-bases.md) | Pipe-scoped AI knowledge bases: list, plain text CRUD, read-access probe | +| [`tools/knowledge-bases.md`](tools/knowledge-bases.md) | Pipe-scoped AI knowledge bases: list, plain text and document (PDF) CRUD, read-access probe | | [`tools/observability.md`](tools/observability.md) | Logs, usage, credits, execution metrics, job exports | | [`tools/members-email-webhooks.md`](tools/members-email-webhooks.md) | Membership, inbox email, webhooks | | [`tools/organization.md`](tools/organization.md) | Organization metadata | diff --git a/docs/mcp/tools/knowledge-bases.md b/docs/mcp/tools/knowledge-bases.md index c77dca4c..4f6c4470 100644 --- a/docs/mcp/tools/knowledge-bases.md +++ b/docs/mcp/tools/knowledge-bases.md @@ -1,6 +1,6 @@ # Knowledge bases -Pipe-scoped AI knowledge bases: list every item on a pipe, full CRUD for plain-text sources, and a read-access probe. **6 tools.** +Pipe-scoped AI knowledge bases: list every item on a pipe, full CRUD for plain-text and document (PDF) sources, and a read-access probe. **10 tools.** Knowledge bases are the data sources an AI agent draws on. Each item's `id` is what you attach to an agent or behavior via `dataSourceIds` (see [Automations & AI](automations-and-ai.md)): use `get_ai_knowledge_bases` to discover the IDs, then `validate_ai_agent_behaviors(data_source_ids=[...])` to check membership before writing the agent. @@ -13,6 +13,10 @@ Knowledge bases are the data sources an AI agent draws on. Each item's `id` is w | `create_ai_knowledge_base_plain_text` | No | Creates a plain text. `name`, `content` (1-3500 chars), and `description` (1-900 chars) are all required. | | `update_ai_knowledge_base_plain_text` | No | Partial update by `plain_text_id`: pass any of `name` / `content` / `description` (at least one); omitted fields keep their stored value. | | `delete_ai_knowledge_base_plain_text` | No | Deletes a plain text permanently. Two-step: preview with `confirm=false` (default), execute with `confirm=true`. | +| `get_ai_knowledge_base_document` | Yes | Fetches one document by `id`. `content` is the stored document URL, not the extracted text. | +| `create_ai_knowledge_base_document` | No | Creates a document from a local PDF in one shot (presigned URL, S3 PUT, create mutation). `name`, `file_path`, and `description` (1-900 chars) are required. | +| `update_ai_knowledge_base_document` | No | Metadata-only partial update by `document_id`: pass `name` and/or `description` (at least one). The PDF file cannot be replaced. | +| `delete_ai_knowledge_base_document` | No | Deletes a document permanently. Two-step: preview with `confirm=false` (default), execute with `confirm=true`. | | `validate_knowledge_base_access` | Yes | Probes whether the current credential can read a pipe's knowledge bases, classifying failures into structured problems instead of opaque errors. | ## Identifiers: pipe UUID, not numeric ID @@ -29,11 +33,27 @@ Limits fail fast before the network call, so an over-limit value is rejected wit On update, only the fields you pass are validated and sent; the others keep their stored values. +## Documents (one-shot PDF upload) + +`create_ai_knowledge_base_document` takes a local `file_path` and does the whole upload in one call: read the file, request a presigned URL for the pipe's organization (resolved from the pipe UUID), PUT the bytes to S3, then run the create mutation with the persistent download URL. The MCP server reads the file as the user, so `file_path` is a local path (`~` is expanded); the create tool is therefore not exposed in the remote profile. + +Client-side checks fail fast — they are the only guardrail on this path, because the backend skips its own PDF and size validation when a document is supplied as a URL rather than a raw upload: + +- `file_path`: must be a `.pdf` (case-insensitive extension), under 20 MiB. +- `name`: required, non-blank. +- `description`: required, 1-900 characters. As with plain text, the GraphQL schema marks it optional but the backend rejects a blank one. + +Failures are tagged with the step that failed — `file_read`, `presigned_url`, `s3_upload`, or `kb_create` — carried on `error.details.step` (an S3 failure also carries `body_snippet`). + +**Indexing is asynchronous.** A created document is not necessarily searchable by agents the moment the tool returns; vectorization runs in the background. + +**Update is metadata-only.** `update_ai_knowledge_base_document` changes `name` / `description`; it cannot replace the PDF. `content` returned by `get_ai_knowledge_base_document` is the stored document URL, not the extracted text. + ## Probe semantics and the write gate -- A green `validate_knowledge_base_access` proves **read access only** (`read_ai_agents` on the pipe) — never the `manage_ai_agents` entitlement that plain-text create / update / delete require. +- A green `validate_knowledge_base_access` proves **read access only** (`read_ai_agents` on the pipe) — never the `manage_ai_agents` entitlement that plain-text and document create / update / delete require. - An **empty knowledge base list** is a valid green result (`knowledge_base_count: 0`), not a failure. -- The **CLI gates writes** on the probe: `pipefy kb plain-text create` / `update` run the read-access probe first and fail with the classified problem if it is denied, before attempting the mutation. +- The **CLI gates writes** on the probe: `pipefy kb plain-text create` / `update` and `pipefy kb document create` / `update` run the read-access probe first and fail with the classified problem if it is denied, before attempting the mutation. - The **MCP tools stay explicit-validate-first**: create / update do not auto-probe. Call `validate_knowledge_base_access` yourself before writing. - **Deletes require confirmation**: the MCP tool needs `confirm=true`; the CLI needs `--yes` (or an interactive prompt). diff --git a/docs/parity.md b/docs/parity.md index 48040154..52a8d302 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -2,7 +2,7 @@ This matrix is the source of truth for **MCP tool ↔ `pipefy` CLI** coverage. Update it whenever MCP tools or CLI commands are added, renamed, or removed. -**Registry source:** `PIPEFY_TOOL_NAMES` in `packages/mcp/src/pipefy_mcp/tools/registry.py` (must stay in sync with this table: **168** tools). +**Registry source:** `PIPEFY_TOOL_NAMES` in `packages/mcp/src/pipefy_mcp/tools/registry.py` (must stay in sync with this table: **172** tools). **Later CLI coverage:** areas such as attachments, field conditions, email, audit export, traditional automations, exports/usage, introspection, and raw GraphQL appear as **shipped** below when the matching Typer commands exist in `packages/cli`. @@ -30,6 +30,7 @@ For **database records**, `find_records` result nodes may use **`fields`** while | `clone_pipe` | `pipefy pipe clone` | shipped | optional `--org`. | | `create_ai_agent` | `pipefy agent create` | shipped | AI Agents domain; post-v0.1 CLI unless explicitly rescoped. | | `create_ai_automation` | `pipefy ai-automation create` | shipped | AI Automations domain; post-v0.1 CLI unless explicitly rescoped. | +| `create_ai_knowledge_base_document` | `pipefy kb document create` | shipped | Knowledge bases; one-shot PDF upload (presigned URL, S3 PUT, create mutation). Pipe-scoped (`--pipe-uuid`, `--file`, `--name`, `--description` <=900, all required). `.pdf` and 20 MiB cap enforced client-side; step-tagged errors (`file_read`/`presigned_url`/`s3_upload`/`kb_create`). CLI gates on the read-access probe. Indexing is asynchronous. | | `create_ai_knowledge_base_plain_text` | `pipefy kb plain-text create` | shipped | Knowledge bases; pipe-scoped (`--pipe-uuid`, `--name`, `--content` <=3500, `--description` <=900, all required). CLI gates on the read-access probe; limits fail fast client-side. | | `create_automation` | `pipefy automation create` | shipped | (`--pipe`, `--name`, `--trigger-id`, `--action-id`, optional `--extra` JSON). | | `create_card` | `pipefy card create` | shipped | (`--fields` JSON, optional `--title`, optional `--phase-id` for `CreateCardInput.phase_id`). | @@ -54,6 +55,7 @@ For **database records**, `find_records` result nodes may use **`fields`** while | `create_webhook` | `pipefy webhook create` | shipped | — | | `delete_ai_agent` | `pipefy agent delete` | shipped | AI Agents domain. | | `delete_ai_automation` | `pipefy ai-automation delete` | shipped | AI Automations domain. | +| `delete_ai_knowledge_base_document` | `pipefy kb document delete` | shipped | Knowledge bases; pipe-scoped (`--id`, `--pipe-uuid`). MCP requires `confirm=true`; CLI requires `--yes` (or interactive prompt). | | `delete_ai_knowledge_base_plain_text` | `pipefy kb plain-text delete` | shipped | Knowledge bases; pipe-scoped (`--id`, `--pipe-uuid`). MCP requires `confirm=true`; CLI requires `--yes` (or interactive prompt). | | `delete_automation` | `pipefy automation delete` | shipped | destructive: `--yes` or confirm. | | `delete_card` | `pipefy card delete` | shipped | destructive: `--yes` or interactive confirm. | @@ -93,6 +95,7 @@ For **database records**, `find_records` result nodes may use **`fields`** while | `get_ai_automation` | `pipefy ai-automation get` | shipped | AI Automations domain. | | `get_ai_automations` | `pipefy ai-automation list` | shipped | AI Automations domain. | | `get_ai_credit_usage` | `pipefy usage credits` | shipped | (`--organization`, `--period`). | +| `get_ai_knowledge_base_document` | `pipefy kb document get` | shipped | Knowledge bases; pipe-scoped document metadata (`--id`, `--pipe-uuid`); `content` is the stored document URL, not the extracted text. | | `get_ai_knowledge_base_plain_text` | `pipefy kb plain-text get` | shipped | Knowledge bases; pipe-scoped plain text with content (`--id`, `--pipe-uuid`). | | `get_ai_knowledge_bases` | `pipefy kb list` | shipped | Knowledge bases; all items on a pipe (plain text, documents, data lookups) with `id` for `dataSourceIds` (`--pipe-uuid`; no pagination). | | `get_automation` | `pipefy automation get` | shipped | — | @@ -165,6 +168,7 @@ For **database records**, `find_records` result nodes may use **`fields`** while | `unpublish_sub_portal` | `pipefy portal sub-portal unpublish` | shipped | internal_api `updateSubPortalElement(subPortalUuid: null)`; sets `subPortals[].published` to false. | | `update_ai_agent` | `pipefy agent update` | shipped | AI Agents domain. | | `update_ai_automation` | `pipefy ai-automation update` | shipped | AI Automations domain. | +| `update_ai_knowledge_base_document` | `pipefy kb document update` | shipped | Knowledge bases; metadata-only partial update (`--id`, `--pipe-uuid`, optional `--name`/`--description`, at least one); no file replacement. CLI gates on the read-access probe. | | `update_ai_knowledge_base_plain_text` | `pipefy kb plain-text update` | shipped | Knowledge bases; partial update (`--id`, `--pipe-uuid`, optional `--name`/`--content`/`--description`, at least one). CLI gates on the read-access probe; limits fail fast client-side. | | `update_automation` | `pipefy automation update` | shipped | (`--extra` JSON). | | `update_card` | `pipefy card update` | shipped | (`--field-updates` JSON, optional title/labels/assignees/due-date). | @@ -205,6 +209,6 @@ for n in m.body: print(len(v.args[0].elts))" ``` -Expect **168** tool names in `PIPEFY_TOOL_NAMES` and **168** data rows in the parity table (excluding the header rows). +Expect **172** tool names in `PIPEFY_TOOL_NAMES` and **172** data rows in the parity table (excluding the header rows). When adding or removing an MCP tool, update **this file** and `PIPEFY_TOOL_NAMES` in the same change set. diff --git a/packages/cli/src/pipefy_cli/commands/knowledge_base.py b/packages/cli/src/pipefy_cli/commands/knowledge_base.py index 657aeda2..2c6f3303 100644 --- a/packages/cli/src/pipefy_cli/commands/knowledge_base.py +++ b/packages/cli/src/pipefy_cli/commands/knowledge_base.py @@ -1,9 +1,11 @@ -"""AI knowledge bases (pipe-scoped): list, plain text CRUD, and access probe.""" +"""AI knowledge bases (pipe-scoped): list, plain text/document CRUD, and access probe.""" from __future__ import annotations +from pathlib import Path + import typer -from pipefy_sdk import PipefyClient +from pipefy_sdk import KnowledgeBaseDocumentUploadError, PipefyClient from pipefy_cli.commands._common import ( confirm_destructive, @@ -11,15 +13,34 @@ ) kb_app = typer.Typer( - help="AI knowledge bases (pipe-scoped: list, plain text CRUD, access probe).", + help="AI knowledge bases (pipe-scoped: list, plain text/document CRUD, access probe).", no_args_is_help=True, ) plain_text_app = typer.Typer(help="Knowledge base plain texts.", no_args_is_help=True) kb_app.add_typer(plain_text_app, name="plain-text") +document_app = typer.Typer( + help="Knowledge base documents (one-shot PDF upload).", no_args_is_help=True +) +kb_app.add_typer(document_app, name="document") _PIPE_UUID_HELP = "Pipe UUID (not the numeric ID; `pipefy pipe get` shows the uuid)." +def _kb_not_found(kind: str, resource_id: str) -> dict[str, object]: + """Failure payload for a per-id get whose id resolved to nothing. + + Shared by every pipe-scoped knowledge base kind (plain text, document, + data lookup) so the not-found envelope and discovery hint stay identical. + """ + return { + "success": False, + "error": ( + f"{kind} not found: {resource_id}. " + "Use `pipefy kb list` to list knowledge base IDs for the pipe." + ), + } + + @kb_app.command("list") def kb_list( ctx: typer.Context, @@ -85,13 +106,7 @@ async def factory(client: PipefyClient): plain_text_id, pipe_uuid ) if not plain_text: - return { - "success": False, - "error": ( - f"Knowledge base plain text not found: {plain_text_id}. " - "Use `pipefy kb list` to list knowledge base IDs for the pipe." - ), - } + return _kb_not_found("Knowledge base plain text", plain_text_id) return {"success": True, "knowledge_base_plain_text": plain_text} run_cli_command(ctx, json_out, factory, exit_1_on_unsuccessful=True) @@ -199,6 +214,143 @@ async def factory(client: PipefyClient): run_cli_command(ctx, json_out, factory, exit_1_on_unsuccessful=True) +@document_app.command("get") +def kb_document_get( + ctx: typer.Context, + document_id: str = typer.Option( + ..., "--id", help="Knowledge base document ID (from `pipefy kb list`)." + ), + pipe_uuid: str = typer.Option(..., "--pipe-uuid", help=_PIPE_UUID_HELP), + json_out: bool = typer.Option( + False, "--json", "-j", help="Print machine-readable JSON to stdout." + ), +) -> None: + """Fetch one knowledge base document (``get_ai_knowledge_base_document``). + + ``content`` is the stored document URL, not the extracted text. Exits 1 + with ``success: false`` when the API resolves no document for the id + (mirrors the MCP tool's not-found handling). + """ + + async def factory(client: PipefyClient): + document = await client.get_ai_knowledge_base_document(document_id, pipe_uuid) + if not document: + return _kb_not_found("Knowledge base document", document_id) + return {"success": True, "knowledge_base_document": document} + + run_cli_command(ctx, json_out, factory, exit_1_on_unsuccessful=True) + + +@document_app.command("create") +def kb_document_create( + ctx: typer.Context, + pipe_uuid: str = typer.Option(..., "--pipe-uuid", help=_PIPE_UUID_HELP), + file: Path = typer.Option( + ..., + "--file", + "-f", + help="Local .pdf file to upload (<=20 MiB). Supports ~ expansion.", + ), + name: str = typer.Option(..., "--name", help="Display name (required)."), + description: str = typer.Option( + ..., "--description", help="Description (required, 1-900 characters)." + ), + json_out: bool = typer.Option( + False, "--json", "-j", help="Print machine-readable JSON to stdout." + ), +) -> None: + """Create a knowledge base document from a local PDF (``create_ai_knowledge_base_document``). + + One-shot: presigned URL, S3 PUT, then the create mutation. Gated on the + read-access probe. Client-side checks fail fast (``.pdf`` extension, 20 MiB + cap, description 1-900). Indexing is asynchronous, so the document may not be + searchable by agents immediately. + """ + + async def factory(client: PipefyClient): + gate = await _probe_gate(client, pipe_uuid) + if gate is not None: + return gate + try: + document = await client.create_ai_knowledge_base_document( + pipe_uuid, name=name, description=description, file_path=file + ) + except KnowledgeBaseDocumentUploadError as exc: + if exc.step == "file_read": + message = str(exc.__cause__) if exc.__cause__ else str(exc) + raise typer.BadParameter(message) from exc + out: dict[str, object] = { + "success": False, + "step": exc.step, + "message": str(exc), + } + if exc.body_snippet is not None: + out["body_snippet"] = exc.body_snippet + return out + return {"success": True, "knowledge_base_document": document} + + run_cli_command(ctx, json_out, factory, exit_1_on_unsuccessful=True) + + +@document_app.command("update") +def kb_document_update( + ctx: typer.Context, + document_id: str = typer.Option( + ..., "--id", help="Knowledge base document ID to update." + ), + pipe_uuid: str = typer.Option(..., "--pipe-uuid", help=_PIPE_UUID_HELP), + name: str | None = typer.Option(None, "--name", help="New name (non-blank)."), + description: str | None = typer.Option( + None, "--description", help="New description (1-900 characters)." + ), + json_out: bool = typer.Option( + False, "--json", "-j", help="Print machine-readable JSON to stdout." + ), +) -> None: + """Update a knowledge base document's metadata (``update_ai_knowledge_base_document``). + + Metadata only (name/description); the PDF cannot be replaced. Provide at + least one of --name or --description. Gated on the read-access probe. + """ + + async def factory(client: PipefyClient): + gate = await _probe_gate(client, pipe_uuid) + if gate is not None: + return gate + document = await client.update_ai_knowledge_base_document( + document_id, pipe_uuid, name=name, description=description + ) + return {"success": True, "knowledge_base_document": document} + + run_cli_command(ctx, json_out, factory, exit_1_on_unsuccessful=True) + + +@document_app.command("delete") +def kb_document_delete( + ctx: typer.Context, + document_id: str = typer.Option( + ..., "--id", help="Knowledge base document ID to delete." + ), + pipe_uuid: str = typer.Option(..., "--pipe-uuid", help=_PIPE_UUID_HELP), + yes: bool = typer.Option( + False, "--yes", "-y", help="Skip the confirmation prompt." + ), + json_out: bool = typer.Option( + False, "--json", "-j", help="Print machine-readable JSON to stdout." + ), +) -> None: + """Delete a knowledge base document permanently (``delete_ai_knowledge_base_document``).""" + confirm_destructive( + yes=yes, + description=f"knowledge base document {document_id}", + ) + + async def factory(client: PipefyClient): + return await client.delete_ai_knowledge_base_document(document_id, pipe_uuid) + + run_cli_command(ctx, json_out, factory, exit_1_on_unsuccessful=True) + + async def _probe_gate(client: PipefyClient, pipe_uuid: str) -> dict | None: """Gate a write on the read-access probe; return a failure dict or None. diff --git a/packages/cli/tests/fixtures/cli_help_golden.txt b/packages/cli/tests/fixtures/cli_help_golden.txt index b9cb4f4d..8fa96b8a 100644 --- a/packages/cli/tests/fixtures/cli_help_golden.txt +++ b/packages/cli/tests/fixtures/cli_help_golden.txt @@ -58,8 +58,8 @@ Usage: pipefy [OPTIONS] COMMAND [ARGS]... │ graphql Execute arbitrary GraphQL (mutations require --yes). See │ │ docs/cli/self-healing.md. │ │ introspect Discover GraphQL types and operations (JSON default). │ -│ kb AI knowledge bases (pipe-scoped: list, plain text CRUD, │ -│ access probe). │ +│ kb AI knowledge bases (pipe-scoped: list, plain text/document │ +│ CRUD, access probe). │ │ export Bulk exports (automation jobs). │ │ org Organization operations. │ │ report-org Organization reports. │ @@ -1652,7 +1652,8 @@ Usage: pipefy introspect type [OPTIONS] NAME ### HELP kb Usage: pipefy kb [OPTIONS] COMMAND [ARGS]... - AI knowledge bases (pipe-scoped: list, plain text CRUD, access probe). + AI knowledge bases (pipe-scoped: list, plain text/document CRUD, access + probe). ╭─ Options ────────────────────────────────────────────────────────────────────╮ │ --help Show this message and exit. │ @@ -1663,6 +1664,109 @@ Usage: pipefy kb [OPTIONS] COMMAND [ARGS]... │ validate-access Probe knowledge-base read access │ │ (``validate_knowledge_base_access``). │ │ plain-text Knowledge base plain texts. │ +│ document Knowledge base documents (one-shot PDF upload). │ +╰──────────────────────────────────────────────────────────────────────────────╯ + +### HELP kb/document +Usage: pipefy kb document [OPTIONS] COMMAND [ARGS]... + + Knowledge base documents (one-shot PDF upload). + +╭─ Options ────────────────────────────────────────────────────────────────────╮ +│ --help Show this message and exit. │ +╰──────────────────────────────────────────────────────────────────────────────╯ +╭─ Commands ───────────────────────────────────────────────────────────────────╮ +│ get Fetch one knowledge base document │ +│ (``get_ai_knowledge_base_document``). │ +│ create Create a knowledge base document from a local PDF │ +│ (``create_ai_knowledge_base_document``). │ +│ update Update a knowledge base document's metadata │ +│ (``update_ai_knowledge_base_document``). │ +│ delete Delete a knowledge base document permanently │ +│ (``delete_ai_knowledge_base_document``). │ +╰──────────────────────────────────────────────────────────────────────────────╯ + +### HELP kb/document/create +Usage: pipefy kb document create [OPTIONS] + + Create a knowledge base document from a local PDF + (``create_ai_knowledge_base_document``). + + One-shot: presigned URL, S3 PUT, then the create mutation. Gated on the + read-access probe. Client-side checks fail fast (``.pdf`` extension, 20 MiB + cap, description 1-900). Indexing is asynchronous, so the document may not be + searchable by agents immediately. + +╭─ Options ────────────────────────────────────────────────────────────────────╮ +│ * --pipe-uuid TEXT Pipe UUID (not the numeric ID; `pipefy pipe │ +│ get` shows the uuid). │ +│ [required] │ +│ * --file -f PATH Local .pdf file to upload (<=20 MiB). │ +│ Supports ~ expansion. │ +│ [required] │ +│ * --name TEXT Display name (required). [required] │ +│ * --description TEXT Description (required, 1-900 characters). │ +│ [required] │ +│ --json -j Print machine-readable JSON to stdout. │ +│ --help Show this message and exit. │ +╰──────────────────────────────────────────────────────────────────────────────╯ + +### HELP kb/document/delete +Usage: pipefy kb document delete [OPTIONS] + + Delete a knowledge base document permanently + (``delete_ai_knowledge_base_document``). + +╭─ Options ────────────────────────────────────────────────────────────────────╮ +│ * --id TEXT Knowledge base document ID to delete. │ +│ [required] │ +│ * --pipe-uuid TEXT Pipe UUID (not the numeric ID; `pipefy pipe │ +│ get` shows the uuid). │ +│ [required] │ +│ --yes -y Skip the confirmation prompt. │ +│ --json -j Print machine-readable JSON to stdout. │ +│ --help Show this message and exit. │ +╰──────────────────────────────────────────────────────────────────────────────╯ + +### HELP kb/document/get +Usage: pipefy kb document get [OPTIONS] + + Fetch one knowledge base document (``get_ai_knowledge_base_document``). + + ``content`` is the stored document URL, not the extracted text. Exits 1 with + ``success: false`` when the API resolves no document for the id (mirrors the + MCP tool's not-found handling). + +╭─ Options ────────────────────────────────────────────────────────────────────╮ +│ * --id TEXT Knowledge base document ID (from `pipefy kb │ +│ list`). │ +│ [required] │ +│ * --pipe-uuid TEXT Pipe UUID (not the numeric ID; `pipefy pipe │ +│ get` shows the uuid). │ +│ [required] │ +│ --json -j Print machine-readable JSON to stdout. │ +│ --help Show this message and exit. │ +╰──────────────────────────────────────────────────────────────────────────────╯ + +### HELP kb/document/update +Usage: pipefy kb document update [OPTIONS] + + Update a knowledge base document's metadata + (``update_ai_knowledge_base_document``). + + Metadata only (name/description); the PDF cannot be replaced. Provide at least + one of --name or --description. Gated on the read-access probe. + +╭─ Options ────────────────────────────────────────────────────────────────────╮ +│ * --id TEXT Knowledge base document ID to update. │ +│ [required] │ +│ * --pipe-uuid TEXT Pipe UUID (not the numeric ID; `pipefy pipe │ +│ get` shows the uuid). │ +│ [required] │ +│ --name TEXT New name (non-blank). │ +│ --description TEXT New description (1-900 characters). │ +│ --json -j Print machine-readable JSON to stdout. │ +│ --help Show this message and exit. │ ╰──────────────────────────────────────────────────────────────────────────────╯ ### HELP kb/list diff --git a/packages/cli/tests/test_knowledge_base_commands.py b/packages/cli/tests/test_knowledge_base_commands.py index 374718cf..b554df97 100644 --- a/packages/cli/tests/test_knowledge_base_commands.py +++ b/packages/cli/tests/test_knowledge_base_commands.py @@ -302,3 +302,271 @@ def test_kb_plain_text_delete_aborts_without_confirmation( ) assert result.exit_code != 0 mock_client.delete_ai_knowledge_base_plain_text.assert_not_awaited() + + +DOCUMENT_FULL = { + "id": "kb-2", + "name": "Handbook", + "description": "Company handbook", + "content": "https://app.pipefy.com/storage/v1/signed/orgs/o/u/h.pdf?sig=x", + "updatedAt": "2026-07-16T00:00:00Z", +} + + +def _pdf(tmp_path): + path = tmp_path / "handbook.pdf" + path.write_bytes(b"%PDF-1.4 body") + return path + + +def test_kb_document_get_json(runner, clean_pipefy_env, saved_cwd, monkeypatch): + _env(monkeypatch) + mock_client = MagicMock() + mock_client.get_ai_knowledge_base_document = AsyncMock(return_value=DOCUMENT_FULL) + + with _client_patch(mock_client): + result = runner.invoke( + app, + ["kb", "document", "get", "--id", "kb-2", "--pipe-uuid", "p", "--json"], + ) + assert result.exit_code == 0, result.stdout + (result.stderr or "") + data = json.loads(result.stdout) + assert data["success"] is True + assert data["knowledge_base_document"] == DOCUMENT_FULL + mock_client.get_ai_knowledge_base_document.assert_awaited_once_with("kb-2", "p") + + +def test_kb_document_get_empty_result_exits_1( + runner, clean_pipefy_env, saved_cwd, monkeypatch +): + """Missing id resolves to `{}` from the SDK; the CLI must not report success.""" + _env(monkeypatch) + mock_client = MagicMock() + mock_client.get_ai_knowledge_base_document = AsyncMock(return_value={}) + + with _client_patch(mock_client): + result = runner.invoke( + app, + [ + "kb", + "document", + "get", + "--id", + "kb-missing", + "--pipe-uuid", + "pipe-uuid-1", + "--json", + ], + ) + assert result.exit_code == 1 + data = json.loads(result.stdout) + assert data["success"] is False + assert "not found" in data["error"].lower() + + +def test_kb_document_create_gated_success( + runner, clean_pipefy_env, saved_cwd, monkeypatch, tmp_path +): + _env(monkeypatch) + pdf = _pdf(tmp_path) + mock_client = MagicMock() + mock_client.validate_knowledge_base_access = AsyncMock(return_value=GREEN_PROBE) + mock_client.create_ai_knowledge_base_document = AsyncMock( + return_value=DOCUMENT_FULL + ) + + with _client_patch(mock_client): + result = runner.invoke( + app, + [ + "kb", + "document", + "create", + "--pipe-uuid", + "pipe-uuid-1", + "--file", + str(pdf), + "--name", + "Handbook", + "--description", + "Company handbook", + "--json", + ], + ) + assert result.exit_code == 0, result.stdout + (result.stderr or "") + assert json.loads(result.stdout)["success"] is True + mock_client.create_ai_knowledge_base_document.assert_awaited_once_with( + "pipe-uuid-1", + name="Handbook", + description="Company handbook", + file_path=pdf, + ) + + +def test_kb_document_create_denied_probe_blocks_write( + runner, clean_pipefy_env, saved_cwd, monkeypatch, tmp_path +): + _env(monkeypatch) + pdf = _pdf(tmp_path) + mock_client = MagicMock() + mock_client.validate_knowledge_base_access = AsyncMock(return_value=DENIED_PROBE) + mock_client.create_ai_knowledge_base_document = AsyncMock() + + with _client_patch(mock_client): + result = runner.invoke( + app, + [ + "kb", + "document", + "create", + "--pipe-uuid", + "pipe-uuid-1", + "--file", + str(pdf), + "--name", + "n", + "--description", + "d", + "--json", + ], + ) + assert result.exit_code == 1 + assert json.loads(result.stdout)["success"] is False + mock_client.create_ai_knowledge_base_document.assert_not_awaited() + + +def test_kb_document_create_s3_step_error_json( + runner, clean_pipefy_env, saved_cwd, monkeypatch, tmp_path +): + from pipefy_sdk import KnowledgeBaseDocumentUploadError + + _env(monkeypatch) + pdf = _pdf(tmp_path) + mock_client = MagicMock() + mock_client.validate_knowledge_base_access = AsyncMock(return_value=GREEN_PROBE) + mock_client.create_ai_knowledge_base_document = AsyncMock( + side_effect=KnowledgeBaseDocumentUploadError( + "S3 upload failed with HTTP 403.", + step="s3_upload", + body_snippet="AccessDenied", + status_code=403, + ) + ) + + with _client_patch(mock_client): + result = runner.invoke( + app, + [ + "kb", + "document", + "create", + "--pipe-uuid", + "pipe-uuid-1", + "--file", + str(pdf), + "--name", + "n", + "--description", + "d", + "--json", + ], + ) + assert result.exit_code == 1 + data = json.loads(result.stdout) + assert data["success"] is False + assert data["step"] == "s3_upload" + assert data["body_snippet"] == "AccessDenied" + + +def test_kb_document_create_file_read_error_is_bad_parameter( + runner, clean_pipefy_env, saved_cwd, monkeypatch, tmp_path +): + from pipefy_sdk import KnowledgeBaseDocumentUploadError + + _env(monkeypatch) + pdf = _pdf(tmp_path) + mock_client = MagicMock() + mock_client.validate_knowledge_base_access = AsyncMock(return_value=GREEN_PROBE) + mock_client.create_ai_knowledge_base_document = AsyncMock( + side_effect=KnowledgeBaseDocumentUploadError( + "File must be a .pdf: notes.txt", step="file_read" + ) + ) + + with _client_patch(mock_client): + result = runner.invoke( + app, + [ + "kb", + "document", + "create", + "--pipe-uuid", + "pipe-uuid-1", + "--file", + str(pdf), + "--name", + "n", + "--description", + "d", + ], + ) + assert result.exit_code == 2 + assert ".pdf" in (result.stderr or result.stdout) + + +def test_kb_document_update_partial(runner, clean_pipefy_env, saved_cwd, monkeypatch): + _env(monkeypatch) + mock_client = MagicMock() + mock_client.validate_knowledge_base_access = AsyncMock(return_value=GREEN_PROBE) + mock_client.update_ai_knowledge_base_document = AsyncMock( + return_value=DOCUMENT_FULL + ) + + with _client_patch(mock_client): + result = runner.invoke( + app, + [ + "kb", + "document", + "update", + "--id", + "kb-2", + "--pipe-uuid", + "pipe-uuid-1", + "--name", + "New name", + "--json", + ], + ) + assert result.exit_code == 0, result.stdout + (result.stderr or "") + mock_client.update_ai_knowledge_base_document.assert_awaited_once_with( + "kb-2", "pipe-uuid-1", name="New name", description=None + ) + + +def test_kb_document_delete_with_yes(runner, clean_pipefy_env, saved_cwd, monkeypatch): + _env(monkeypatch) + mock_client = MagicMock() + mock_client.delete_ai_knowledge_base_document = AsyncMock( + return_value={"success": True, "errors": []} + ) + + with _client_patch(mock_client): + result = runner.invoke( + app, + [ + "kb", + "document", + "delete", + "--id", + "kb-2", + "--pipe-uuid", + "pipe-uuid-1", + "--yes", + "--json", + ], + ) + assert result.exit_code == 0, result.stdout + (result.stderr or "") + mock_client.delete_ai_knowledge_base_document.assert_awaited_once_with( + "kb-2", "pipe-uuid-1" + ) diff --git a/packages/mcp/src/pipefy_mcp/tools/knowledge_base_tools.py b/packages/mcp/src/pipefy_mcp/tools/knowledge_base_tools.py index eed022b6..3795b443 100644 --- a/packages/mcp/src/pipefy_mcp/tools/knowledge_base_tools.py +++ b/packages/mcp/src/pipefy_mcp/tools/knowledge_base_tools.py @@ -12,7 +12,7 @@ from mcp.server.fastmcp import Context, FastMCP from mcp.types import ToolAnnotations -from pipefy_sdk import classify_exception +from pipefy_sdk import KnowledgeBaseDocumentUploadError, classify_exception from pipefy_mcp.core.tool_error_envelope import ( is_unified_envelope_enabled, @@ -27,6 +27,33 @@ ) +def _kb_document_upload_error( + exc: KnowledgeBaseDocumentUploadError, +) -> dict[str, Any]: + """Map a step-tagged document upload failure onto the tool failure envelope. + + ``file_read`` and ``s3_upload`` failures carry their own actionable message; + ``presigned_url``/``kb_create`` failures are GraphQL/transport errors, so the + shared SDK classifier is used for a clean kind/code (falling back to the raw + message). ``step`` (and any S3 ``body_snippet``) always rides in ``details``. + """ + details: dict[str, Any] = {"step": exc.step} + if exc.step == "file_read": + message = str(exc.__cause__) if exc.__cause__ else str(exc) + return tool_error(message, details=details) + if exc.step == "s3_upload": + if exc.body_snippet: + details["body_snippet"] = exc.body_snippet + return tool_error(str(exc), details=details) + problem = classify_exception(exc.__cause__) if exc.__cause__ else None + if problem is None: + return tool_error(str(exc), details=details) + details["kind"] = problem.kind.value + if problem.correlation_id: + details["correlation_id"] = problem.correlation_id + return tool_error(problem.message, code=problem.code, details=details) + + def _kb_tool_error_from_exception( exc: BaseException, *, not_found_hint: bool = True ) -> dict[str, Any]: @@ -280,6 +307,196 @@ async def delete_ai_knowledge_base_plain_text( message="Knowledge base plain text deleted.", ) + @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True)) + async def get_ai_knowledge_base_document( + ctx: Context, + document_id: str, + pipe_uuid: str, + ) -> dict[str, Any]: + """Fetch one pipe-scoped knowledge base document by ID. + + `content` is the stored document URL (where the PDF was uploaded), + not the extracted text. + + Args: + document_id: Knowledge base document ID (from `get_ai_knowledge_bases`). + pipe_uuid: Pipe UUID (not the numeric ID). + """ + client = get_pipefy_client(ctx) + err = _blank_error(document_id, "document_id") or _blank_error( + pipe_uuid, "pipe_uuid" + ) + if err is not None: + return err + try: + document = await client.get_ai_knowledge_base_document( + document_id.strip(), pipe_uuid.strip() + ) + except Exception as exc: # noqa: BLE001 + return _kb_tool_error_from_exception(exc) + if not document: + return tool_error( + f"Knowledge base document not found: {document_id.strip()}. " + f"{_KB_ID_DISCOVERY_HINT}" + ) + return _kb_success( + {"knowledge_base_document": document}, + message="Knowledge base document retrieved.", + ) + + # Left unmarked for the remote profile: create reads a local file_path, + # which has no meaning on a hosted server (mirrors the attachment tools). + @mcp.tool( + annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False), + ) + async def create_ai_knowledge_base_document( + ctx: Context, + pipe_uuid: str, + name: str, + description: str, + file_path: str, + ) -> dict[str, Any]: + """Create a pipe-scoped knowledge base document from a local PDF (one-shot upload). Requires manage_ai_agents on the pipe; run `validate_knowledge_base_access` first. + + Reads the local PDF the MCP server (running as the user) can access, + uploads it via a presigned URL, then registers the document. `~` is + expanded. Client-side checks fail fast: the file must be a `.pdf` + (case-insensitive) under 20 MiB, and `description` is 1-900 + characters (required). Indexing is asynchronous: the document may + not be searchable by agents immediately. To attach it to an agent, + pass the returned `id` in the agent's or behavior's `dataSourceIds`. + + Args: + pipe_uuid: Pipe UUID (not the numeric ID). + name: Display name (required, non-blank). + description: Description (required, 1-900 characters). + file_path: Local path to a `.pdf` file. Supports `~` expansion. + """ + client = get_pipefy_client(ctx) + err = _blank_error(pipe_uuid, "pipe_uuid") or _blank_error( + file_path, "file_path" + ) + if err is not None: + return err + try: + document = await client.create_ai_knowledge_base_document( + pipe_uuid.strip(), + name=name, + description=description, + file_path=file_path.strip(), + ) + except KnowledgeBaseDocumentUploadError as exc: + return _kb_document_upload_error(exc) + except ValueError as exc: + return tool_error(str(exc)) + except Exception as exc: # noqa: BLE001 + return _kb_tool_error_from_exception(exc) + return _kb_success( + {"knowledge_base_document": document}, + message="Knowledge base document created.", + ) + + @mcp.tool( + annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False), + ) + async def update_ai_knowledge_base_document( + ctx: Context, + document_id: str, + pipe_uuid: str, + name: str | None = None, + description: str | None = None, + ) -> dict[str, Any]: + """Update a knowledge base document's metadata. Requires manage_ai_agents; run `validate_knowledge_base_access` first. + + Metadata only: `name` and/or `description`; the PDF file cannot be + replaced. Provide at least one field. When given, `description` is + 1-900 characters (enforced client-side). + + Args: + document_id: Document ID to update (from `get_ai_knowledge_bases`). + pipe_uuid: Pipe UUID (not the numeric ID). + name: New name (non-blank when given). + description: New description (1-900 characters when given). + """ + client = get_pipefy_client(ctx) + err = _blank_error(document_id, "document_id") or _blank_error( + pipe_uuid, "pipe_uuid" + ) + if err is not None: + return err + try: + document = await client.update_ai_knowledge_base_document( + document_id.strip(), + pipe_uuid.strip(), + name=name, + description=description, + ) + except ValueError as exc: + return tool_error(str(exc)) + except Exception as exc: # noqa: BLE001 + return _kb_tool_error_from_exception(exc) + return _kb_success( + {"knowledge_base_document": document}, + message="Knowledge base document updated.", + ) + + @mcp.tool( + annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=True), + ) + async def delete_ai_knowledge_base_document( + ctx: Context, + document_id: str, + pipe_uuid: str, + confirm: bool = False, + ) -> dict[str, Any]: + """Delete a pipe-scoped knowledge base document permanently. This action is irreversible. + + Two-step operation: preview with `confirm=False` (default), then + execute with `confirm=True` after explicit human approval. + Elicitation does not authorize deletion (only `confirm=True` does). + Requires manage_ai_agents on the pipe. + + Args: + document_id: Document ID to delete (from `get_ai_knowledge_bases`). + pipe_uuid: Pipe UUID (not the numeric ID). + confirm: Must be `True` to run the delete mutation. + """ + client = get_pipefy_client(ctx) + err = _blank_error(document_id, "document_id") or _blank_error( + pipe_uuid, "pipe_uuid" + ) + if err is not None: + return err + + guard = await check_destructive_confirmation( + ctx, + confirm=confirm, + resource_descriptor=( + f"knowledge base document (ID: {document_id.strip()})" + ), + ) + if guard is not None: + return guard + + try: + result = await client.delete_ai_knowledge_base_document( + document_id.strip(), pipe_uuid.strip() + ) + except Exception as exc: # noqa: BLE001 + return _kb_tool_error_from_exception(exc) + if not result.get("success"): + errs = result.get("errors") or [] + detail = ( + "; ".join(str(e) for e in errs) + if errs + else "API returned success=false" + ) + return tool_error(f"delete_ai_knowledge_base_document failed: {detail}") + return _kb_success( + {"deleted_id": document_id.strip()}, + message="Knowledge base document deleted.", + ) + @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True)) async def validate_knowledge_base_access( ctx: Context, diff --git a/packages/mcp/src/pipefy_mcp/tools/registry.py b/packages/mcp/src/pipefy_mcp/tools/registry.py index 51d03134..ff373087 100644 --- a/packages/mcp/src/pipefy_mcp/tools/registry.py +++ b/packages/mcp/src/pipefy_mcp/tools/registry.py @@ -44,6 +44,7 @@ "create_automation", "create_card", "create_card_relation", + "create_ai_knowledge_base_document", "create_ai_knowledge_base_plain_text", "create_field_condition", "create_ipaas_connection", @@ -65,6 +66,7 @@ "create_webhook", "delete_ai_agent", "delete_ai_automation", + "delete_ai_knowledge_base_document", "delete_ai_knowledge_base_plain_text", "delete_automation", "delete_card", @@ -104,6 +106,7 @@ "get_ai_automation", "get_ai_automations", "get_ai_credit_usage", + "get_ai_knowledge_base_document", "get_ai_knowledge_base_plain_text", "get_ai_knowledge_bases", "get_automation", @@ -176,6 +179,7 @@ "unpublish_sub_portal", "update_ai_agent", "update_ai_automation", + "update_ai_knowledge_base_document", "update_ai_knowledge_base_plain_text", "update_automation", "update_card", diff --git a/packages/mcp/tests/tools/test_knowledge_base_tools.py b/packages/mcp/tests/tools/test_knowledge_base_tools.py index ba1d7e1f..5ba95c9a 100644 --- a/packages/mcp/tests/tools/test_knowledge_base_tools.py +++ b/packages/mcp/tests/tools/test_knowledge_base_tools.py @@ -8,7 +8,7 @@ from mcp.shared.memory import ( create_connected_server_and_client_session as create_client_session, ) -from pipefy_sdk import PipefyClient +from pipefy_sdk import KnowledgeBaseDocumentUploadError, PipefyClient from pipefy_mcp.core.tool_error_envelope import tool_error_message from pipefy_mcp.tools.knowledge_base_tools import KnowledgeBaseTools @@ -30,6 +30,14 @@ "updatedAt": "2026-07-17T00:00:00Z", } +DOCUMENT_FULL = { + "id": "kb-2", + "name": "Handbook", + "description": "Company handbook", + "content": "https://app.pipefy.com/storage/v1/signed/orgs/o/u/h.pdf?sig=x", + "updatedAt": "2026-07-16T00:00:00Z", +} + def permission_denied_error() -> TransportQueryError: return TransportQueryError( @@ -63,6 +71,10 @@ def mock_kb_client(): client.create_ai_knowledge_base_plain_text = AsyncMock() client.update_ai_knowledge_base_plain_text = AsyncMock() client.delete_ai_knowledge_base_plain_text = AsyncMock() + client.get_ai_knowledge_base_document = AsyncMock() + client.create_ai_knowledge_base_document = AsyncMock() + client.update_ai_knowledge_base_document = AsyncMock() + client.delete_ai_knowledge_base_document = AsyncMock() client.validate_knowledge_base_access = AsyncMock() return client @@ -341,3 +353,229 @@ async def test_get_knowledge_bases_not_found_has_no_self_referential_hint( assert payload["success"] is False assert payload["error"]["details"]["kind"] == "not_found" assert "get_ai_knowledge_bases" not in tool_error_message(payload) + + +async def test_get_document_success( + kb_session, mock_kb_client, unified_envelope, extract_payload +): + mock_kb_client.get_ai_knowledge_base_document = AsyncMock( + return_value=DOCUMENT_FULL + ) + async with kb_session as session: + result = await session.call_tool( + "get_ai_knowledge_base_document", + {"document_id": "kb-2", "pipe_uuid": "pipe-uuid-1"}, + ) + assert result.isError is False + mock_kb_client.get_ai_knowledge_base_document.assert_awaited_once_with( + "kb-2", "pipe-uuid-1" + ) + payload = extract_payload(result) + assert payload["success"] is True + assert payload["data"]["knowledge_base_document"] == DOCUMENT_FULL + + +@pytest.mark.anyio +@pytest.mark.parametrize("kb_session", [None], indirect=True) +async def test_get_document_not_found(kb_session, mock_kb_client, extract_payload): + mock_kb_client.get_ai_knowledge_base_document = AsyncMock(return_value={}) + async with kb_session as session: + result = await session.call_tool( + "get_ai_knowledge_base_document", + {"document_id": "kb-x", "pipe_uuid": "pipe-uuid-1"}, + ) + payload = extract_payload(result) + assert payload["success"] is False + assert "not found" in tool_error_message(payload).lower() + + +@pytest.mark.anyio +@pytest.mark.parametrize("kb_session", [None], indirect=True) +async def test_create_document_success( + kb_session, mock_kb_client, unified_envelope, extract_payload +): + mock_kb_client.create_ai_knowledge_base_document = AsyncMock( + return_value=DOCUMENT_FULL + ) + async with kb_session as session: + result = await session.call_tool( + "create_ai_knowledge_base_document", + { + "pipe_uuid": "pipe-uuid-1", + "name": "Handbook", + "description": "Company handbook", + "file_path": "/tmp/handbook.pdf", + }, + ) + assert result.isError is False + mock_kb_client.create_ai_knowledge_base_document.assert_awaited_once_with( + "pipe-uuid-1", + name="Handbook", + description="Company handbook", + file_path="/tmp/handbook.pdf", + ) + payload = extract_payload(result) + assert payload["success"] is True + assert payload["data"]["knowledge_base_document"] == DOCUMENT_FULL + + +@pytest.mark.anyio +@pytest.mark.parametrize("kb_session", [None], indirect=True) +async def test_create_document_blank_file_path_rejected( + kb_session, mock_kb_client, extract_payload +): + async with kb_session as session: + result = await session.call_tool( + "create_ai_knowledge_base_document", + { + "pipe_uuid": "pipe-uuid-1", + "name": "Handbook", + "description": "d", + "file_path": " ", + }, + ) + payload = extract_payload(result) + assert payload["success"] is False + assert "file_path" in tool_error_message(payload) + mock_kb_client.create_ai_knowledge_base_document.assert_not_awaited() + + +@pytest.mark.anyio +@pytest.mark.parametrize("kb_session", [None], indirect=True) +async def test_create_document_non_pdf_maps_file_read_step( + kb_session, mock_kb_client, extract_payload +): + mock_kb_client.create_ai_knowledge_base_document = AsyncMock( + side_effect=KnowledgeBaseDocumentUploadError( + "File must be a .pdf: notes.txt", step="file_read" + ) + ) + async with kb_session as session: + result = await session.call_tool( + "create_ai_knowledge_base_document", + { + "pipe_uuid": "pipe-uuid-1", + "name": "n", + "description": "d", + "file_path": "/tmp/notes.txt", + }, + ) + payload = extract_payload(result) + assert payload["success"] is False + assert ".pdf" in tool_error_message(payload) + assert payload["error"]["details"]["step"] == "file_read" + + +@pytest.mark.anyio +@pytest.mark.parametrize("kb_session", [None], indirect=True) +async def test_create_document_s3_failure_carries_step_and_snippet( + kb_session, mock_kb_client, extract_payload +): + mock_kb_client.create_ai_knowledge_base_document = AsyncMock( + side_effect=KnowledgeBaseDocumentUploadError( + "S3 upload failed with HTTP 403.", + step="s3_upload", + body_snippet="AccessDenied", + status_code=403, + ) + ) + async with kb_session as session: + result = await session.call_tool( + "create_ai_knowledge_base_document", + { + "pipe_uuid": "pipe-uuid-1", + "name": "n", + "description": "d", + "file_path": "/tmp/handbook.pdf", + }, + ) + payload = extract_payload(result) + assert payload["success"] is False + assert payload["error"]["details"]["step"] == "s3_upload" + assert payload["error"]["details"]["body_snippet"] == "AccessDenied" + + +@pytest.mark.anyio +@pytest.mark.parametrize("kb_session", [None], indirect=True) +async def test_create_document_kb_create_failure_classified( + kb_session, mock_kb_client, extract_payload +): + mock_kb_client.create_ai_knowledge_base_document = AsyncMock( + side_effect=KnowledgeBaseDocumentUploadError( + "Document create failed: denied", + step="kb_create", + ) + ) + mock_kb_client.create_ai_knowledge_base_document.side_effect.__cause__ = ( + permission_denied_error() + ) + async with kb_session as session: + result = await session.call_tool( + "create_ai_knowledge_base_document", + { + "pipe_uuid": "pipe-uuid-1", + "name": "n", + "description": "d", + "file_path": "/tmp/handbook.pdf", + }, + ) + payload = extract_payload(result) + assert payload["success"] is False + assert payload["error"]["details"]["step"] == "kb_create" + assert payload["error"]["code"] == "PERMISSION_DENIED" + + +@pytest.mark.anyio +@pytest.mark.parametrize("kb_session", [None], indirect=True) +async def test_update_document_success( + kb_session, mock_kb_client, unified_envelope, extract_payload +): + mock_kb_client.update_ai_knowledge_base_document = AsyncMock( + return_value=DOCUMENT_FULL + ) + async with kb_session as session: + result = await session.call_tool( + "update_ai_knowledge_base_document", + {"document_id": "kb-2", "pipe_uuid": "pipe-uuid-1", "name": "New name"}, + ) + assert result.isError is False + mock_kb_client.update_ai_knowledge_base_document.assert_awaited_once_with( + "kb-2", "pipe-uuid-1", name="New name", description=None + ) + + +@pytest.mark.anyio +@pytest.mark.parametrize("kb_session", [None], indirect=True) +async def test_delete_document_preview_without_confirm( + kb_session, mock_kb_client, extract_payload +): + async with kb_session as session: + result = await session.call_tool( + "delete_ai_knowledge_base_document", + {"document_id": "kb-2", "pipe_uuid": "pipe-uuid-1"}, + ) + payload = extract_payload(result) + assert payload["success"] is False + assert payload["requires_confirmation"] is True + mock_kb_client.delete_ai_knowledge_base_document.assert_not_awaited() + + +@pytest.mark.anyio +@pytest.mark.parametrize("kb_session", [None], indirect=True) +async def test_delete_document_with_confirm_executes( + kb_session, mock_kb_client, unified_envelope, extract_payload +): + mock_kb_client.delete_ai_knowledge_base_document = AsyncMock( + return_value={"success": True, "errors": []} + ) + async with kb_session as session: + result = await session.call_tool( + "delete_ai_knowledge_base_document", + {"document_id": "kb-2", "pipe_uuid": "pipe-uuid-1", "confirm": True}, + ) + assert result.isError is False + payload = extract_payload(result) + assert payload["success"] is True + mock_kb_client.delete_ai_knowledge_base_document.assert_awaited_once_with( + "kb-2", "pipe-uuid-1" + ) diff --git a/packages/sdk/src/pipefy_sdk/__init__.py b/packages/sdk/src/pipefy_sdk/__init__.py index 77fe47ba..7d4b2685 100644 --- a/packages/sdk/src/pipefy_sdk/__init__.py +++ b/packages/sdk/src/pipefy_sdk/__init__.py @@ -72,6 +72,8 @@ CardSearch, KnowledgeBaseAccessProbeResult, KnowledgeBaseDeleteResult, + KnowledgeBaseDocumentPayload, + KnowledgeBaseDocumentUploadError, KnowledgeBasePayload, KnowledgeBasePlainTextPayload, LlmProviderPayload, @@ -118,6 +120,8 @@ "GraphQLProblemKind", "KnowledgeBaseAccessProbeResult", "KnowledgeBaseDeleteResult", + "KnowledgeBaseDocumentPayload", + "KnowledgeBaseDocumentUploadError", "KnowledgeBasePayload", "KnowledgeBasePlainTextPayload", "LlmProviderPayload", diff --git a/packages/sdk/src/pipefy_sdk/client.py b/packages/sdk/src/pipefy_sdk/client.py index 6a1ceb9d..2f26a16c 100644 --- a/packages/sdk/src/pipefy_sdk/client.py +++ b/packages/sdk/src/pipefy_sdk/client.py @@ -3,6 +3,7 @@ from __future__ import annotations from dataclasses import dataclass +from pathlib import Path from typing import Any from httpx import Auth @@ -83,6 +84,7 @@ CardSearch, KnowledgeBaseAccessProbeResult, KnowledgeBaseDeleteResult, + KnowledgeBaseDocumentPayload, KnowledgeBasePayload, KnowledgeBasePlainTextPayload, LlmProviderPayload, @@ -1024,6 +1026,53 @@ async def delete_ai_knowledge_base_plain_text( plain_text_id, pipe_uuid ) + async def get_ai_knowledge_base_document( + self, document_id: str, pipe_uuid: str + ) -> KnowledgeBaseDocumentPayload: + """Fetch one pipe-scoped knowledge base document by id.""" + return await self._knowledge_base_service.get_ai_knowledge_base_document( + document_id, pipe_uuid + ) + + async def create_ai_knowledge_base_document( + self, + pipe_uuid: str, + *, + name: str, + description: str, + file_path: str | Path, + ) -> KnowledgeBaseDocumentPayload: + """Create a pipe-scoped knowledge base document from a local PDF (one-shot upload). + + Reads the PDF (``.pdf`` and 20 MiB cap enforced client-side), uploads it + via a presigned URL, then runs the create mutation. Raises + ``KnowledgeBaseDocumentUploadError`` (with ``step``) on pipeline failure. + """ + return await self._knowledge_base_service.create_ai_knowledge_base_document( + pipe_uuid, name=name, description=description, file_path=file_path + ) + + async def update_ai_knowledge_base_document( + self, + document_id: str, + pipe_uuid: str, + *, + name: str | None = None, + description: str | None = None, + ) -> KnowledgeBaseDocumentPayload: + """Update a knowledge base document's metadata (name/description; no file replacement).""" + return await self._knowledge_base_service.update_ai_knowledge_base_document( + document_id, pipe_uuid, name=name, description=description + ) + + async def delete_ai_knowledge_base_document( + self, document_id: str, pipe_uuid: str + ) -> KnowledgeBaseDeleteResult: + """Delete a pipe-scoped knowledge base document (permanent).""" + return await self._knowledge_base_service.delete_ai_knowledge_base_document( + document_id, pipe_uuid + ) + async def validate_knowledge_base_access( self, pipe_uuid: str ) -> KnowledgeBaseAccessProbeResult: diff --git a/packages/sdk/src/pipefy_sdk/queries/knowledge_base_queries.py b/packages/sdk/src/pipefy_sdk/queries/knowledge_base_queries.py index a2363d5f..c0c540ab 100644 --- a/packages/sdk/src/pipefy_sdk/queries/knowledge_base_queries.py +++ b/packages/sdk/src/pipefy_sdk/queries/knowledge_base_queries.py @@ -1,10 +1,16 @@ -"""GraphQL operations for pipe-scoped AI knowledge bases (list + plain text CRUD). +"""GraphQL operations for pipe-scoped AI knowledge bases (list, plain text, documents). All operations are pipe-scoped by ``pipeUuid`` (the pipe UUID, not the numeric id). The list query returns every knowledge base item on a pipe (plain text, documents, data lookups) as a plain ``[AiKnowledgeBase]`` list — there is no -Relay connection and no pagination. Plain-text create/update/delete are Relay -mutations that take a single ``input`` object. +Relay connection and no pagination. Plain-text and document create/update/delete +are Relay mutations that take a single ``input`` object. + +Documents are created from an already-uploaded PDF: the ``documentUrl`` is the +persistent download URL returned by the ``createPresignedUrl`` mutation (in +``attachment_queries``) after the bytes are PUT to its presigned upload URL. The +presigned request needs the pipe's organization, resolved from the pipe UUID via +:data:`GET_PIPE_ORGANIZATION_QUERY`. """ from __future__ import annotations @@ -91,10 +97,95 @@ """ ) +# Resolves the organization that owns a pipe from the pipe UUID. The document +# upload flow needs an organization id for the presigned URL, but the tools are +# pipe-scoped, so the id is derived here rather than asked of the caller. The +# ``pipe`` root field accepts the pipe UUID as its ``id`` argument. +GET_PIPE_ORGANIZATION_QUERY = gql( + """ + query pipeOrganization($id: ID!) { + pipe(id: $id) { + organization { + id + uuid + } + } + } + """ +) + +GET_AI_KNOWLEDGE_BASE_DOCUMENT_QUERY = gql( + """ + query aiKnowledgeBaseDocument($id: ID!, $pipeUuid: ID!) { + aiKnowledgeBaseDocument(id: $id, pipeUuid: $pipeUuid) { + id + name + description + content + updatedAt + } + } + """ +) + +CREATE_AI_KNOWLEDGE_BASE_DOCUMENT_MUTATION = gql( + """ + mutation createAiKnowledgeBaseDocument( + $input: CreateKnowledgeBaseDocumentInput! + ) { + createAiKnowledgeBaseDocument(input: $input) { + knowledgeBaseDocument { + id + name + description + content + updatedAt + } + } + } + """ +) + +UPDATE_AI_KNOWLEDGE_BASE_DOCUMENT_MUTATION = gql( + """ + mutation updateAiKnowledgeBaseDocument( + $input: UpdateKnowledgeBaseDocumentInput! + ) { + updateAiKnowledgeBaseDocument(input: $input) { + knowledgeBaseDocument { + id + name + description + content + updatedAt + } + } + } + """ +) + +DELETE_AI_KNOWLEDGE_BASE_DOCUMENT_MUTATION = gql( + """ + mutation deleteAiKnowledgeBaseDocument( + $input: DeleteKnowledgeBaseDocumentInput! + ) { + deleteAiKnowledgeBaseDocument(input: $input) { + success + errors + } + } + """ +) + __all__ = [ + "CREATE_AI_KNOWLEDGE_BASE_DOCUMENT_MUTATION", "CREATE_AI_KNOWLEDGE_BASE_PLAIN_TEXT_MUTATION", + "DELETE_AI_KNOWLEDGE_BASE_DOCUMENT_MUTATION", "DELETE_AI_KNOWLEDGE_BASE_PLAIN_TEXT_MUTATION", "GET_AI_KNOWLEDGE_BASES_QUERY", + "GET_AI_KNOWLEDGE_BASE_DOCUMENT_QUERY", "GET_AI_KNOWLEDGE_BASE_PLAIN_TEXT_QUERY", + "GET_PIPE_ORGANIZATION_QUERY", + "UPDATE_AI_KNOWLEDGE_BASE_DOCUMENT_MUTATION", "UPDATE_AI_KNOWLEDGE_BASE_PLAIN_TEXT_MUTATION", ] diff --git a/packages/sdk/src/pipefy_sdk/services/knowledge_base_service.py b/packages/sdk/src/pipefy_sdk/services/knowledge_base_service.py index db22428b..3d0123be 100644 --- a/packages/sdk/src/pipefy_sdk/services/knowledge_base_service.py +++ b/packages/sdk/src/pipefy_sdk/services/knowledge_base_service.py @@ -1,4 +1,4 @@ -"""Service for pipe-scoped AI knowledge bases: list, plain text CRUD, and probe. +"""Service for pipe-scoped AI knowledge bases: list, plain text/document CRUD, probe. Client-side limits fail fast before the network call so callers get an actionable ``ValueError`` instead of a backend 422. The limits mirror the @@ -7,28 +7,53 @@ ``description`` is required (the GraphQL schema marks it optional, but the backend rejects a blank one) and capped at 900 chars. Updates omit any field the caller leaves unset, but validate every field they do pass. + +Documents are created one-shot from a local PDF: read the file (``.pdf`` +extension and 20 MiB cap enforced here, because the backend skips both when the +document arrives as a URL rather than a raw upload), mint a presigned URL, +PUT the bytes, then run the create mutation with the persistent download URL. +Every stage raises :class:`KnowledgeBaseDocumentUploadError` tagged with the +step that failed. The presign and S3 PUT primitives are shared with the +attachment upload pipeline. """ from __future__ import annotations +import asyncio +from pathlib import Path from typing import Any +from pipefy_infra.filesystem import LocalFile, LocalFileError + from pipefy_sdk.graphql_executor import GraphQLExecutor from pipefy_sdk.graphql_problem import ( GraphQLProblem, GraphQLProblemKind, classify_graphql_error_dicts, ) +from pipefy_sdk.queries.attachment_queries import CREATE_PRESIGNED_URL_MUTATION from pipefy_sdk.queries.knowledge_base_queries import ( + CREATE_AI_KNOWLEDGE_BASE_DOCUMENT_MUTATION, CREATE_AI_KNOWLEDGE_BASE_PLAIN_TEXT_MUTATION, + DELETE_AI_KNOWLEDGE_BASE_DOCUMENT_MUTATION, DELETE_AI_KNOWLEDGE_BASE_PLAIN_TEXT_MUTATION, + GET_AI_KNOWLEDGE_BASE_DOCUMENT_QUERY, GET_AI_KNOWLEDGE_BASE_PLAIN_TEXT_QUERY, GET_AI_KNOWLEDGE_BASES_QUERY, + GET_PIPE_ORGANIZATION_QUERY, + UPDATE_AI_KNOWLEDGE_BASE_DOCUMENT_MUTATION, UPDATE_AI_KNOWLEDGE_BASE_PLAIN_TEXT_MUTATION, ) +from pipefy_sdk.services.attachment_service import ( + _ALLOWED_UPLOAD_HOST_RE, + HttpxS3Uploader, + S3Uploader, +) from pipefy_sdk.services.types import ( KnowledgeBaseAccessProbeResult, KnowledgeBaseDeleteResult, + KnowledgeBaseDocumentPayload, + KnowledgeBaseDocumentUploadError, KnowledgeBasePayload, KnowledgeBasePlainTextPayload, ) @@ -36,6 +61,14 @@ MAX_PLAIN_TEXT_CONTENT_LENGTH = 3500 MAX_PLAIN_TEXT_DESCRIPTION_LENGTH = 900 +# Document policy, enforced client-side because the backend skips PDF/size +# validation when a document arrives as a URL (the only path these tools use). +# The description cap matches the shared ``DataSource`` rule (max 900). +DOCUMENT_PDF_SUFFIX = ".pdf" +MAX_DOCUMENT_SIZE_BYTES = 20 * 1024 * 1024 +MAX_DOCUMENT_DESCRIPTION_LENGTH = 900 +DOCUMENT_CONTENT_TYPE = "application/pdf" + _PROBE_READ_ONLY_NOTE = ( "Read access confirmed. This proves knowledge-base list/read access only " "(read_ai_agents on the pipe), never write entitlement; plain-text " @@ -73,10 +106,18 @@ def _problem_dict(problem: GraphQLProblem) -> dict[str, Any]: class KnowledgeBaseService: - """Pipe-scoped AI knowledge base reads, plain-text writes, and the probe.""" + """Pipe-scoped AI knowledge base reads, plain-text/document writes, and the probe.""" - def __init__(self, *, executor: GraphQLExecutor) -> None: + def __init__( + self, + *, + executor: GraphQLExecutor, + s3_uploader: S3Uploader | None = None, + ) -> None: self._executor = executor + self._s3_uploader: S3Uploader = s3_uploader or HttpxS3Uploader( + allowed_host_pattern=_ALLOWED_UPLOAD_HOST_RE + ) async def get_ai_knowledge_bases( self, pipe_uuid: str @@ -226,6 +267,253 @@ async def delete_ai_knowledge_base_plain_text( "errors": [str(e) for e in errors] if isinstance(errors, list) else [], } + async def get_ai_knowledge_base_document( + self, document_id: str, pipe_uuid: str + ) -> KnowledgeBaseDocumentPayload: + """Fetch one pipe-scoped knowledge base document by id. + + Args: + document_id: Knowledge base document ID (from the list). + pipe_uuid: Pipe UUID (not the numeric id). + + Returns: + The document dict; empty dict when the API resolves nothing. + ``content`` is the stored document URL, not the extracted text. + """ + variables = { + "id": _require_non_blank(document_id, "document_id"), + "pipeUuid": _require_non_blank(pipe_uuid, "pipe_uuid"), + } + response = await self._executor.execute_query( + GET_AI_KNOWLEDGE_BASE_DOCUMENT_QUERY, variables + ) + document = response.get("aiKnowledgeBaseDocument") + return document if isinstance(document, dict) else {} + + async def create_ai_knowledge_base_document( + self, + pipe_uuid: str, + *, + name: str, + description: str, + file_path: str | Path, + ) -> KnowledgeBaseDocumentPayload: + """Create a pipe-scoped knowledge base document from a local PDF (one-shot). + + The pipeline reads the local PDF (validating the ``.pdf`` extension and + the 20 MiB cap client-side, since the backend skips both on the URL + path), requests a presigned URL for the pipe's organization, PUTs the + bytes, and runs the create mutation with the persistent download URL. + Indexing is asynchronous: the document may not be searchable by agents + immediately after this returns. + + Args: + pipe_uuid: Pipe UUID (not the numeric id). + name: Display name (required, non-blank). + description: Description (required by the backend, 1-900 chars). + file_path: Local path to a ``.pdf`` file (``~`` is expanded). + + Returns: + The created document dict. + + Raises: + ValueError: When ``name``/``description``/``pipe_uuid`` are invalid. + KnowledgeBaseDocumentUploadError: On any pipeline failure (``step`` + identifies the stage: ``file_read``/``presigned_url``/ + ``s3_upload``/``kb_create``). + """ + pipe_uuid = _require_non_blank(pipe_uuid, "pipe_uuid") + name = _require_non_blank(name, "name") + description = _require_bounded( + description, "description", MAX_DOCUMENT_DESCRIPTION_LENGTH + ) + + file = await self._read_pdf(file_path) + + download_url = await self._upload_pdf(pipe_uuid, file) + + input_obj = { + "pipeUuid": pipe_uuid, + "name": name, + "description": description, + "documentUrl": download_url, + } + try: + response = await self._executor.execute_query( + CREATE_AI_KNOWLEDGE_BASE_DOCUMENT_MUTATION, {"input": input_obj} + ) + return _unwrap_document(response, "createAiKnowledgeBaseDocument") + except Exception as exc: + raise KnowledgeBaseDocumentUploadError( + f"Document create failed: {exc}", step="kb_create" + ) from exc + + async def update_ai_knowledge_base_document( + self, + document_id: str, + pipe_uuid: str, + *, + name: str | None = None, + description: str | None = None, + ) -> KnowledgeBaseDocumentPayload: + """Update a pipe-scoped knowledge base document (metadata only). + + Only ``name``/``description`` can change; file replacement is not + supported. Only the fields you pass are sent; unset fields keep their + stored value. At least one of ``name``/``description`` must be provided. + + Args: + document_id: Document ID to update. + pipe_uuid: Pipe UUID (not the numeric id). + name: New name (non-blank when given). + description: New description (1-900 chars when given). + + Returns: + The updated document dict. + """ + if name is None and description is None: + raise ValueError("Provide at least one of name or description to update.") + input_obj: dict[str, Any] = { + "pipeUuid": _require_non_blank(pipe_uuid, "pipe_uuid"), + "documentId": _require_non_blank(document_id, "document_id"), + } + if name is not None: + input_obj["name"] = _require_non_blank(name, "name") + if description is not None: + input_obj["description"] = _require_bounded( + description, "description", MAX_DOCUMENT_DESCRIPTION_LENGTH + ) + response = await self._executor.execute_query( + UPDATE_AI_KNOWLEDGE_BASE_DOCUMENT_MUTATION, {"input": input_obj} + ) + return _unwrap_document(response, "updateAiKnowledgeBaseDocument") + + async def delete_ai_knowledge_base_document( + self, document_id: str, pipe_uuid: str + ) -> KnowledgeBaseDeleteResult: + """Delete a pipe-scoped knowledge base document (permanent). + + Args: + document_id: Document ID to delete. + pipe_uuid: Pipe UUID (not the numeric id). + + Returns: + ``success`` and any backend ``errors``. + """ + input_obj = { + "pipeUuid": _require_non_blank(pipe_uuid, "pipe_uuid"), + "documentId": _require_non_blank(document_id, "document_id"), + } + response = await self._executor.execute_query( + DELETE_AI_KNOWLEDGE_BASE_DOCUMENT_MUTATION, {"input": input_obj} + ) + payload = response.get("deleteAiKnowledgeBaseDocument") + payload = payload if isinstance(payload, dict) else {} + errors = payload.get("errors") + return { + "success": bool(payload.get("success")), + "errors": [str(e) for e in errors] if isinstance(errors, list) else [], + } + + async def _read_pdf(self, file_path: str | Path) -> LocalFile: + """Read and validate the local PDF (``file_read`` step).""" + if Path(file_path).suffix.lower() != DOCUMENT_PDF_SUFFIX: + raise KnowledgeBaseDocumentUploadError( + f"File must be a .pdf: {file_path}", step="file_read" + ) + file = LocalFile(Path(file_path), max_size_bytes=MAX_DOCUMENT_SIZE_BYTES) + try: + await asyncio.to_thread(file.read) + except LocalFileError as exc: + raise KnowledgeBaseDocumentUploadError(str(exc), step="file_read") from exc + return file + + async def _upload_pdf(self, pipe_uuid: str, file: LocalFile) -> str: + """Resolve the org, presign, and PUT the bytes; return the download URL. + + Covers the ``presigned_url`` and ``s3_upload`` steps. + """ + try: + organization_id = await self._resolve_organization_id(pipe_uuid) + presigned = await self._create_presigned_url( + organization_id, file.name, file.size + ) + except KnowledgeBaseDocumentUploadError: + raise + except Exception as exc: + raise KnowledgeBaseDocumentUploadError( + f"Presigned URL request failed: {exc}", step="presigned_url" + ) from exc + + upload_url = presigned.get("url") + download_url = presigned.get("download_url") + if not isinstance(upload_url, str) or not upload_url.strip(): + raise KnowledgeBaseDocumentUploadError( + "Pipefy did not return a presigned upload URL.", + step="presigned_url", + ) + if not isinstance(download_url, str) or not download_url.strip(): + raise KnowledgeBaseDocumentUploadError( + "Pipefy did not return a document download URL.", + step="presigned_url", + ) + + try: + put_result = await self._s3_uploader.put( + url=upload_url.strip(), + bytes_=file.bytes, + content_type=DOCUMENT_CONTENT_TYPE, + ) + except Exception as exc: + # Transport errors and the uploader's host-allowlist rejection are + # s3_upload-stage failures; tag them so the step contract holds. + raise KnowledgeBaseDocumentUploadError( + f"S3 upload failed: {exc}", step="s3_upload" + ) from exc + status = put_result.get("status_code", 0) + if not isinstance(status, int) or status >= 400: + body_snippet = put_result.get("body_snippet") + raise KnowledgeBaseDocumentUploadError( + f"S3 upload failed with HTTP {status}.", + step="s3_upload", + body_snippet=body_snippet if isinstance(body_snippet, str) else None, + status_code=status if isinstance(status, int) else None, + ) + return download_url.strip() + + async def _resolve_organization_id(self, pipe_uuid: str) -> str: + """Resolve the organization id for a pipe (presign needs it).""" + response = await self._executor.execute_query( + GET_PIPE_ORGANIZATION_QUERY, {"id": pipe_uuid} + ) + pipe = response.get("pipe") + organization = pipe.get("organization") if isinstance(pipe, dict) else None + org_id = organization.get("id") if isinstance(organization, dict) else None + if not isinstance(org_id, str) or not org_id.strip(): + raise KnowledgeBaseDocumentUploadError( + f"Could not resolve the organization for pipe {pipe_uuid}.", + step="presigned_url", + ) + return org_id.strip() + + async def _create_presigned_url( + self, organization_id: str, file_name: str, content_length: int + ) -> dict[str, Any]: + """Request a presigned upload URL from Pipefy (shared mutation).""" + payload = await self._executor.execute_query( + CREATE_PRESIGNED_URL_MUTATION, + { + "organizationId": organization_id, + "fileName": file_name, + "contentType": DOCUMENT_CONTENT_TYPE, + "contentLength": content_length, + }, + ) + node = payload.get("createPresignedUrl") + if not isinstance(node, dict): + return {"url": None, "download_url": None} + return {"url": node.get("url"), "download_url": node.get("downloadUrl")} + async def validate_knowledge_base_access( self, pipe_uuid: str ) -> KnowledgeBaseAccessProbeResult: @@ -284,3 +572,22 @@ def _unwrap_plain_text( f"{mutation_key} returned no plain text payload; " "the write may not have persisted." ) + + +def _unwrap_document( + response: dict[str, Any], mutation_key: str +) -> KnowledgeBaseDocumentPayload: + """Unwrap a write mutation's document; a missing payload is a failure. + + A write that returns no GraphQL errors but a null ``knowledgeBaseDocument`` + must not read as success — the caller cannot know whether it persisted. + """ + payload = response.get(mutation_key) + if isinstance(payload, dict): + document = payload.get("knowledgeBaseDocument") + if isinstance(document, dict) and document: + return document + raise ValueError( + f"{mutation_key} returned no document payload; " + "the write may not have persisted." + ) diff --git a/packages/sdk/src/pipefy_sdk/services/types.py b/packages/sdk/src/pipefy_sdk/services/types.py index e8f79491..3cb3d7ab 100644 --- a/packages/sdk/src/pipefy_sdk/services/types.py +++ b/packages/sdk/src/pipefy_sdk/services/types.py @@ -166,6 +166,55 @@ class KnowledgeBaseDeleteResult(TypedDict): errors: list[str] +class KnowledgeBaseDocumentPayload(TypedDict, total=False): + """A single pipe-scoped knowledge base document. + + Returned by the get query and by create/update. ``content`` is the stored + document URL (the persistent download URL the file was uploaded to), not the + extracted text; it may be null on legacy rows. ``description`` is required on + write (the backing ``DataSource`` model enforces presence) but may be null on + legacy rows. + """ + + id: str + name: str + description: str | None + content: str | None + updatedAt: str | None + + +KnowledgeBaseDocumentUploadStep = Literal[ + "file_read", + "presigned_url", + "s3_upload", + "kb_create", +] + + +class KnowledgeBaseDocumentUploadError(Exception): + """Raised on knowledge base document upload pipeline failure. + + The ``step`` attribute identifies which stage failed (``file_read`` reads and + validates the local PDF, ``presigned_url`` resolves the org and mints the + upload URL, ``s3_upload`` PUTs the bytes, ``kb_create`` runs the create + mutation) so surfaces can map it to a step-aware error envelope (MCP) or + typer message (CLI) without parsing strings. + """ + + def __init__( + self, + message: str, + *, + step: KnowledgeBaseDocumentUploadStep, + body_snippet: str | None = None, + status_code: int | None = None, + ) -> None: + self.step = step + self.body_snippet = body_snippet + self.status_code = status_code + super().__init__(message) + + class KnowledgeBaseAccessProbeResult(TypedDict, total=False): """Outcome of the knowledge-base read-access probe (pipe-scoped). diff --git a/packages/sdk/tests/services/test_knowledge_base_service.py b/packages/sdk/tests/services/test_knowledge_base_service.py index a3df6d53..1a47f082 100644 --- a/packages/sdk/tests/services/test_knowledge_base_service.py +++ b/packages/sdk/tests/services/test_knowledge_base_service.py @@ -6,11 +6,61 @@ from _shared.mock_clients import mock_executor from pipefy_sdk.graphql_executor import GraphQLResult +from pipefy_sdk.services import knowledge_base_service as kb_module from pipefy_sdk.services.knowledge_base_service import ( + MAX_DOCUMENT_DESCRIPTION_LENGTH, MAX_PLAIN_TEXT_CONTENT_LENGTH, MAX_PLAIN_TEXT_DESCRIPTION_LENGTH, KnowledgeBaseService, ) +from pipefy_sdk.services.types import KnowledgeBaseDocumentUploadError + +DOCUMENT_FULL = { + "id": "kb-2", + "name": "Handbook", + "description": "Company handbook", + "content": "https://app.pipefy.com/storage/v1/signed/orgs/o/uploads/u/h.pdf?sig=x", + "updatedAt": "2026-07-16T00:00:00Z", +} + +_UPLOAD_URL = "https://pipefy-prd.s3.amazonaws.com/orgs/o/uploads/u/h.pdf?X-Amz=1" +_DOWNLOAD_URL = ( + "https://app.pipefy.com/storage/v1/signed/orgs/o/uploads/u/h.pdf?signature=x" +) +_PDF_BYTES = b"%PDF-1.4 fake pdf body" + + +class _FakeUploader: + """S3Uploader test double: records the PUT and returns a canned result.""" + + def __init__(self, result: dict | None = None, error: Exception | None = None): + self.result = result if result is not None else {"status_code": 200} + self.error = error + self.calls: list[dict] = [] + + async def put(self, *, url: str, bytes_: bytes, content_type: str | None) -> dict: + self.calls.append({"url": url, "bytes_": bytes_, "content_type": content_type}) + if self.error is not None: + raise self.error + return self.result + + +def _write_pdf(tmp_path, name: str = "handbook.pdf", data: bytes = _PDF_BYTES): + path = tmp_path / name + path.write_bytes(data) + return path + + +def _create_flow_executor(): + """Executor whose three ``execute_query`` calls drive a full create flow.""" + return mock_executor( + side_effect=[ + {"pipe": {"organization": {"id": "300514213", "uuid": "org-uuid"}}}, + {"createPresignedUrl": {"url": _UPLOAD_URL, "downloadUrl": _DOWNLOAD_URL}}, + {"createAiKnowledgeBaseDocument": {"knowledgeBaseDocument": DOCUMENT_FULL}}, + ] + ) + PLAIN_TEXT_NODE = { "id": "kb-1", @@ -379,3 +429,330 @@ async def test_null_data_without_errors_reports_runtime_problem(self): assert probe["ok"] is False assert probe["problem"]["kind"] == "runtime" + + +class TestGetDocument: + @pytest.mark.anyio + async def test_returns_document(self): + executor = mock_executor({"aiKnowledgeBaseDocument": DOCUMENT_FULL}) + service = KnowledgeBaseService(executor=executor) + + result = await service.get_ai_knowledge_base_document("kb-2", "pipe-uuid-1") + + assert result == DOCUMENT_FULL + _, variables = executor.execute_query.await_args.args + assert variables == {"id": "kb-2", "pipeUuid": "pipe-uuid-1"} + + @pytest.mark.anyio + async def test_null_yields_empty_dict(self): + executor = mock_executor({"aiKnowledgeBaseDocument": None}) + service = KnowledgeBaseService(executor=executor) + + assert await service.get_ai_knowledge_base_document("kb-x", "p") == {} + + +class TestCreateDocument: + @pytest.mark.anyio + async def test_full_pipeline_sends_expected_wire_calls(self, tmp_path): + executor = _create_flow_executor() + uploader = _FakeUploader() + service = KnowledgeBaseService(executor=executor, s3_uploader=uploader) + pdf = _write_pdf(tmp_path) + + result = await service.create_ai_knowledge_base_document( + "pipe-uuid-1", + name="Handbook", + description="Company handbook", + file_path=pdf, + ) + + assert result == DOCUMENT_FULL + calls = executor.execute_query.await_args_list + # 1) resolve org from pipe uuid + assert calls[0].args[1] == {"id": "pipe-uuid-1"} + # 2) presign with the resolved org id, file name, pdf type, and byte length + assert calls[1].args[1] == { + "organizationId": "300514213", + "fileName": "handbook.pdf", + "contentType": "application/pdf", + "contentLength": len(_PDF_BYTES), + } + # 3) create mutation stores the persistent download URL as documentUrl + assert calls[2].args[1] == { + "input": { + "pipeUuid": "pipe-uuid-1", + "name": "Handbook", + "description": "Company handbook", + "documentUrl": _DOWNLOAD_URL, + } + } + # the bytes were PUT to the single-use upload URL, not the download URL + assert uploader.calls == [ + { + "url": _UPLOAD_URL, + "bytes_": _PDF_BYTES, + "content_type": "application/pdf", + } + ] + + @pytest.mark.anyio + async def test_non_pdf_rejected_at_file_read_before_any_io(self, tmp_path): + executor = _create_flow_executor() + uploader = _FakeUploader() + service = KnowledgeBaseService(executor=executor, s3_uploader=uploader) + txt = _write_pdf(tmp_path, name="notes.txt") + + with pytest.raises(KnowledgeBaseDocumentUploadError) as exc_info: + await service.create_ai_knowledge_base_document( + "p", name="n", description="d", file_path=txt + ) + + assert exc_info.value.step == "file_read" + assert ".pdf" in str(exc_info.value) + executor.execute_query.assert_not_awaited() + assert uploader.calls == [] + + @pytest.mark.anyio + async def test_uppercase_pdf_extension_accepted(self, tmp_path): + executor = _create_flow_executor() + service = KnowledgeBaseService(executor=executor, s3_uploader=_FakeUploader()) + pdf = _write_pdf(tmp_path, name="Handbook.PDF") + + result = await service.create_ai_knowledge_base_document( + "p", name="n", description="d", file_path=pdf + ) + + assert result == DOCUMENT_FULL + + @pytest.mark.anyio + async def test_size_cap_rejected_at_file_read(self, tmp_path, monkeypatch): + monkeypatch.setattr(kb_module, "MAX_DOCUMENT_SIZE_BYTES", 8) + executor = _create_flow_executor() + uploader = _FakeUploader() + service = KnowledgeBaseService(executor=executor, s3_uploader=uploader) + pdf = _write_pdf(tmp_path, data=b"%PDF-1.4 this is well over eight bytes") + + with pytest.raises(KnowledgeBaseDocumentUploadError) as exc_info: + await service.create_ai_knowledge_base_document( + "p", name="n", description="d", file_path=pdf + ) + + assert exc_info.value.step == "file_read" + executor.execute_query.assert_not_awaited() + assert uploader.calls == [] + + @pytest.mark.anyio + @pytest.mark.parametrize("field", ["name", "description"]) + async def test_blank_required_field_rejected_before_io(self, tmp_path, field): + executor = _create_flow_executor() + uploader = _FakeUploader() + service = KnowledgeBaseService(executor=executor, s3_uploader=uploader) + pdf = _write_pdf(tmp_path) + kwargs = {"name": "n", "description": "d", field: " "} + + with pytest.raises(ValueError, match=field): + await service.create_ai_knowledge_base_document( + "p", file_path=pdf, **kwargs + ) + executor.execute_query.assert_not_awaited() + assert uploader.calls == [] + + @pytest.mark.anyio + async def test_description_over_limit_rejected(self, tmp_path): + executor = _create_flow_executor() + service = KnowledgeBaseService(executor=executor, s3_uploader=_FakeUploader()) + pdf = _write_pdf(tmp_path) + + with pytest.raises(ValueError, match="900"): + await service.create_ai_knowledge_base_document( + "p", + name="n", + description="y" * (MAX_DOCUMENT_DESCRIPTION_LENGTH + 1), + file_path=pdf, + ) + executor.execute_query.assert_not_awaited() + + @pytest.mark.anyio + async def test_org_resolution_failure_tagged_presigned_url(self, tmp_path): + executor = mock_executor(side_effect=[{"pipe": None}]) + uploader = _FakeUploader() + service = KnowledgeBaseService(executor=executor, s3_uploader=uploader) + pdf = _write_pdf(tmp_path) + + with pytest.raises(KnowledgeBaseDocumentUploadError) as exc_info: + await service.create_ai_knowledge_base_document( + "p", name="n", description="d", file_path=pdf + ) + + assert exc_info.value.step == "presigned_url" + assert uploader.calls == [] + + @pytest.mark.anyio + async def test_missing_upload_url_tagged_presigned_url(self, tmp_path): + executor = mock_executor( + side_effect=[ + {"pipe": {"organization": {"id": "300514213"}}}, + {"createPresignedUrl": {"url": None, "downloadUrl": _DOWNLOAD_URL}}, + ] + ) + uploader = _FakeUploader() + service = KnowledgeBaseService(executor=executor, s3_uploader=uploader) + pdf = _write_pdf(tmp_path) + + with pytest.raises(KnowledgeBaseDocumentUploadError) as exc_info: + await service.create_ai_knowledge_base_document( + "p", name="n", description="d", file_path=pdf + ) + + assert exc_info.value.step == "presigned_url" + assert uploader.calls == [] + + @pytest.mark.anyio + async def test_s3_failure_tagged_and_carries_snippet(self, tmp_path): + executor = mock_executor( + side_effect=[ + {"pipe": {"organization": {"id": "300514213"}}}, + { + "createPresignedUrl": { + "url": _UPLOAD_URL, + "downloadUrl": _DOWNLOAD_URL, + } + }, + ] + ) + uploader = _FakeUploader( + result={"status_code": 403, "body_snippet": "AccessDenied"} + ) + service = KnowledgeBaseService(executor=executor, s3_uploader=uploader) + pdf = _write_pdf(tmp_path) + + with pytest.raises(KnowledgeBaseDocumentUploadError) as exc_info: + await service.create_ai_knowledge_base_document( + "p", name="n", description="d", file_path=pdf + ) + + assert exc_info.value.step == "s3_upload" + assert exc_info.value.status_code == 403 + assert exc_info.value.body_snippet == "AccessDenied" + # the create mutation (third call) never ran + assert executor.execute_query.await_count == 2 + + @pytest.mark.anyio + async def test_s3_put_exception_tagged_s3_upload(self, tmp_path): + """A raising PUT (transport error, allowlist rejection) carries the step tag.""" + executor = mock_executor( + side_effect=[ + {"pipe": {"organization": {"id": "300514213"}}}, + { + "createPresignedUrl": { + "url": _UPLOAD_URL, + "downloadUrl": _DOWNLOAD_URL, + } + }, + ] + ) + uploader = _FakeUploader(error=ConnectionError("connection reset by peer")) + service = KnowledgeBaseService(executor=executor, s3_uploader=uploader) + pdf = _write_pdf(tmp_path) + + with pytest.raises(KnowledgeBaseDocumentUploadError) as exc_info: + await service.create_ai_knowledge_base_document( + "p", name="n", description="d", file_path=pdf + ) + + assert exc_info.value.step == "s3_upload" + assert "connection reset by peer" in str(exc_info.value) + # the create mutation (third call) never ran + assert executor.execute_query.await_count == 2 + + @pytest.mark.anyio + async def test_create_mutation_failure_tagged_kb_create(self, tmp_path): + executor = mock_executor( + side_effect=[ + {"pipe": {"organization": {"id": "300514213"}}}, + { + "createPresignedUrl": { + "url": _UPLOAD_URL, + "downloadUrl": _DOWNLOAD_URL, + } + }, + RuntimeError("mutation boom"), + ] + ) + service = KnowledgeBaseService(executor=executor, s3_uploader=_FakeUploader()) + pdf = _write_pdf(tmp_path) + + with pytest.raises(KnowledgeBaseDocumentUploadError) as exc_info: + await service.create_ai_knowledge_base_document( + "p", name="n", description="d", file_path=pdf + ) + + assert exc_info.value.step == "kb_create" + + +class TestUpdateDocument: + @pytest.mark.anyio + async def test_only_sends_provided_fields(self): + executor = mock_executor( + {"updateAiKnowledgeBaseDocument": {"knowledgeBaseDocument": DOCUMENT_FULL}} + ) + service = KnowledgeBaseService(executor=executor) + + await service.update_ai_knowledge_base_document( + "kb-2", "pipe-uuid-1", name="New name" + ) + + _, variables = executor.execute_query.await_args.args + assert variables == { + "input": { + "pipeUuid": "pipe-uuid-1", + "documentId": "kb-2", + "name": "New name", + } + } + + @pytest.mark.anyio + async def test_no_fields_rejected_before_wire(self): + executor = mock_executor({}) + service = KnowledgeBaseService(executor=executor) + + with pytest.raises(ValueError, match="at least one"): + await service.update_ai_knowledge_base_document("kb-2", "pipe-uuid-1") + executor.execute_query.assert_not_awaited() + + @pytest.mark.anyio + async def test_description_over_limit_rejected(self): + executor = mock_executor({}) + service = KnowledgeBaseService(executor=executor) + + with pytest.raises(ValueError, match="900"): + await service.update_ai_knowledge_base_document( + "kb-2", "p", description="y" * (MAX_DOCUMENT_DESCRIPTION_LENGTH + 1) + ) + executor.execute_query.assert_not_awaited() + + +class TestDeleteDocument: + @pytest.mark.anyio + async def test_success(self): + executor = mock_executor( + {"deleteAiKnowledgeBaseDocument": {"success": True, "errors": []}} + ) + service = KnowledgeBaseService(executor=executor) + + result = await service.delete_ai_knowledge_base_document("kb-2", "p") + + assert result == {"success": True, "errors": []} + _, variables = executor.execute_query.await_args.args + assert variables == {"input": {"pipeUuid": "p", "documentId": "kb-2"}} + + @pytest.mark.anyio + async def test_failure_surfaces_errors(self): + executor = mock_executor( + {"deleteAiKnowledgeBaseDocument": {"success": False, "errors": ["nope"]}} + ) + service = KnowledgeBaseService(executor=executor) + + result = await service.delete_ai_knowledge_base_document("kb-2", "p") + + assert result == {"success": False, "errors": ["nope"]} diff --git a/skills/ai-agents/pipefy-ai-agents/SKILL.md b/skills/ai-agents/pipefy-ai-agents/SKILL.md index ed15a949..c00765ef 100644 --- a/skills/ai-agents/pipefy-ai-agents/SKILL.md +++ b/skills/ai-agents/pipefy-ai-agents/SKILL.md @@ -4,7 +4,7 @@ description: > Use this skill when the user wants to create, read, update, delete, or troubleshoot AI agents (conversational agents with behaviors). Covers 7 MCP tools including pre-flight validation, plus pipe-scoped - knowledge bases (list, plain text CRUD, access probe) attached via dataSourceIds. + knowledge bases (list, plain text and document CRUD, access probe) attached via dataSourceIds. For traditional automations and AI automations, see skills/automations/. tags: [pipefy, ai-agents, behaviors, conversational] --- @@ -187,6 +187,10 @@ Knowledge bases are pipe-scoped data sources an agent draws on. Attach one by pu | `create_ai_knowledge_base_plain_text` | `pipefy kb plain-text create` | No | Create a plain text (`name`, `content` 1-3500, `description` 1-900 — all required). | | `update_ai_knowledge_base_plain_text` | `pipefy kb plain-text update` | No | Partial update; pass at least one of name/content/description. | | `delete_ai_knowledge_base_plain_text` | `pipefy kb plain-text delete` | No | **(Two-step destructive)** MCP needs `confirm=true`; CLI needs `--yes`. | +| `get_ai_knowledge_base_document` | `pipefy kb document get` | Yes | Fetch one document's metadata (`content` is the stored URL, not text). | +| `create_ai_knowledge_base_document` | `pipefy kb document create` | No | Upload a local PDF in one shot (`file_path`/`--file`, `name`, `description` 1-900). `.pdf` + 20 MiB cap client-side; indexing is async. | +| `update_ai_knowledge_base_document` | `pipefy kb document update` | No | Metadata-only update (name/description); no file replacement. | +| `delete_ai_knowledge_base_document` | `pipefy kb document delete` | No | **(Two-step destructive)** MCP needs `confirm=true`; CLI needs `--yes`. | | `validate_knowledge_base_access` | `pipefy kb validate-access` | Yes | Probe read access before writes. | ### Flow: validate-access → create plain text → attach @@ -195,6 +199,8 @@ Knowledge bases are pipe-scoped data sources an agent draws on. Attach one by pu 2. **Create the source** — `create_ai_knowledge_base_plain_text(pipe_uuid, name, content, description)`. Limits fail fast client-side: `content` 1-3500 chars, `description` 1-900 chars (both required). Keep the returned `id`. 3. **Attach** — add that `id` to a behavior's `dataSourceIds` (or the agent-level `data_source_ids`) when calling `create_ai_agent` / `update_ai_agent`. Validate first with `validate_ai_agent_behaviors(pipe_id, behaviors, data_source_ids=[...])` — unknown IDs surface as warnings. +For a **PDF document** instead of plain text, use `create_ai_knowledge_base_document(pipe_uuid, name, description, file_path)` (CLI: `pipefy kb document create --file …`) at step 2. It uploads the local PDF in one shot; `.pdf` and the 20 MiB cap are enforced client-side, and indexing is asynchronous (the document may not be searchable immediately). The rest of the flow is identical — keep the returned `id` and attach it. + --- ## Token normalization & slug resolution