Artifacts

Artifacts are binary files — PDFs, images, spreadsheets, and other attachments — that capabilities produce or consume during tool invocations. The orchestrator manages the entire artifact lifecycle: capabilities never communicate with each other directly.

This page explains how the artifact system works and how to implement artifact support in your capability.

How it works

There are two directions:

• Output artifacts — a capability generates a file and the orchestrator retrieves it.
• Input artifacts — the orchestrator sends a previously stored file to a capability.

The orchestrator sits in the middle, storing artifacts in memory with the following constraints:

Constraint	Value
Maximum artifact size	5 MB (MAX_ARTIFACT_BYTES)
Time-to-live	6 hours (ARTIFACT_TTL)
gRPC chunk size	256 KB

Each stored artifact is scoped to a user_id and session_id, so artifacts cannot leak across users or sessions.

Output artifacts (capability produces a file)

When a capability generates a file that needs to be passed along, it follows this flow:

1. Capability generates the file. For example, a PDF agent renders a document.

2. Capability stores the file internally and returns an artifact_id in the tool result JSON:

   {"ok": true, "artifact_id": "abc123", "filename": "report.pdf"}

3. Orchestrator calls DownloadOutputArtifact to stream the binary data from the capability in 256 KB chunks.

4. Orchestrator stores the file as a StoredArtifact with user_id, session_id, filename, mime_type, and the raw data.

5. Orchestrator rewrites the tool result that the LLM sees, replacing the raw artifact_id with full metadata:

   {
     "ok": true,
     "artifact": {
       "artifact_id": "abc123",
       "filename": "report.pdf",
       "mime_type": "application/pdf",
       "size_bytes": 245678
     }
   }

6. The LLM can now reference artifact_id in subsequent tool calls — for example, to email the file as an attachment.

gRPC RPC

rpc DownloadOutputArtifact(DownloadOutputArtifactRequest) returns (stream ArtifactChunk);

The orchestrator sends a DownloadOutputArtifactRequest with the artifact_id, and the capability streams back ArtifactChunk messages until done is true.

Input artifacts (capability consumes a file)

When the LLM wants to pass a previously produced file to another capability (or the same one), it includes an attachments array in the tool arguments:

1. LLM calls a tool with an attachments array containing artifact_id references:

   {"to": "alice@example.com", "subject": "Report", "attachments": [{"artifact_id": "abc123"}]}

2. Orchestrator intercepts the tool call in prepare_attachment_artifacts_for_capability.

3. Orchestrator verifies ownership — the user_id and session_id on the stored artifact must match.

4. Orchestrator uploads the artifact to the target capability via UploadInputArtifact gRPC streaming (256 KB chunks).

5. Capability returns a capability_artifact_id — its own internal reference for the uploaded file.

6. Orchestrator rewrites the tool arguments, replacing each artifact_id with the capability’s own reference plus metadata:

   {
     "to": "alice@example.com",
     "subject": "Report",
     "attachments": [{
       "capability_artifact_id": "cap-xyz",
       "filename": "report.pdf",
       "mime_type": "application/pdf",
       "size_bytes": 245678
     }]
   }

gRPC RPC

rpc UploadInputArtifact(stream UploadInputArtifactChunk) returns (UploadInputArtifactResponse);

The orchestrator streams UploadInputArtifactChunk messages to the capability. The capability reassembles the file and returns a UploadInputArtifactResponse containing the capability_artifact_id.

End-to-end example

Here is a concrete example of artifacts flowing through the system:

User: “Create a PDF about marine biology and email it to alice@example.com”

1. The LLM calls pdf__create_pdf_document with content arguments.
2. The PDF capability generates the PDF, stores it internally, and returns:

   {"ok": true, "artifact_id": "abc123", "filename": "marine_biology.pdf"}

3. The orchestrator downloads the PDF via DownloadOutputArtifact and stores it.
4. The orchestrator rewrites the result so the LLM sees:

   {"ok": true, "artifact": {"artifact_id": "abc123", "filename": "marine_biology.pdf", "mime_type": "application/pdf", "size_bytes": 245678}}

5. The LLM calls pim__send_email with:

   {"to": "alice@example.com", "subject": "Marine Biology", "attachments": [{"artifact_id": "abc123"}]}

6. The orchestrator uploads the PDF to the PIM capability via UploadInputArtifact.
7. The PIM capability receives the file, constructs a MIME email with the attachment, and sends it.

Tipp

The PDF capability and the PIM capability never talk to each other. The orchestrator handles all file transfer between them.

Implementing output artifacts

To produce artifacts from your capability, you need to:

1. Store generated files internally, keyed by a unique ID.
2. Return an artifact_id field in your tool result JSON.
3. Implement the DownloadOutputArtifact RPC to stream the data back when requested.

server.py — producing artifacts

import uuid
import json
import grpc
from concurrent import futures
import capability_pb2 as pb2
import capability_pb2_grpc as pb2_grpc

class MyCapability(pb2_grpc.CapabilityServicer):
    def __init__(self):
        self._output_artifacts = {}

    def Invoke(self, request, context):
        args = json.loads(request.args_json)

        # ... generate the file ...
        pdf_bytes = generate_pdf(args)

        artifact_id = str(uuid.uuid4())
        self._output_artifacts[artifact_id] = {
            "data": pdf_bytes,
            "filename": "report.pdf",
            "mime_type": "application/pdf",
        }

        return pb2.InvokeResponse(
            result_json=json.dumps({
                "ok": True,
                "artifact_id": artifact_id,
                "filename": "report.pdf",
            }).encode()
        )

    def DownloadOutputArtifact(self, request, context):
        artifact = self._output_artifacts.get(request.artifact_id)
        if not artifact:
            yield pb2.ArtifactChunk(error="Artifact not found")
            return

        data = artifact["data"]
        chunk_size = 256 * 1024  # 256 KB
        for i in range(0, len(data), chunk_size):
            yield pb2.ArtifactChunk(
                data=data[i:i + chunk_size],
                filename=artifact["filename"],
                mime_type=artifact["mime_type"],
                done=(i + chunk_size >= len(data)),
            )

    def Healthcheck(self, request, context):
        return pb2.HealthResponse(ready=True)

Achtung

The orchestrator enforces a 5 MB maximum per artifact. If your capability produces files that could exceed this limit, check the size before returning an artifact_id and return an error instead.

Implementing input artifacts

To consume artifacts that the orchestrator uploads to your capability:

1. Implement the UploadInputArtifact RPC to receive streamed data.
2. Store uploaded artifacts keyed by the returned capability_artifact_id.
3. In your tool handler, look up files by capability_artifact_id from the rewritten arguments.

server.py — consuming artifacts

class MyCapability(pb2_grpc.CapabilityServicer):
    def __init__(self):
        self._input_artifacts = {}

    def UploadInputArtifact(self, request_iterator, context):
        chunks = []
        filename = ""
        mime_type = ""

        for chunk in request_iterator:
            if chunk.filename:
                filename = chunk.filename
            if chunk.mime_type:
                mime_type = chunk.mime_type
            if chunk.data:
                chunks.append(chunk.data)

        artifact_id = str(uuid.uuid4())
        self._input_artifacts[artifact_id] = {
            "data": b"".join(chunks),
            "filename": filename,
            "mime_type": mime_type,
        }

        return pb2.UploadInputArtifactResponse(
            capability_artifact_id=artifact_id
        )

    def Invoke(self, request, context):
        args = json.loads(request.args_json)

        # The orchestrator has rewritten the attachments array —
        # each entry now has capability_artifact_id instead of artifact_id.
        for attachment in args.get("attachments", []):
            cap_id = attachment["capability_artifact_id"]
            file_data = self._input_artifacts[cap_id]
            # Use file_data["data"], file_data["filename"], etc.

        return pb2.InvokeResponse(
            result_json=json.dumps({"ok": True}).encode()
        )

    def Healthcheck(self, request, context):
        return pb2.HealthResponse(ready=True)

Error scenarios

Scenario	What happens
artifact_id not found on the capability	DownloadOutputArtifact yields an ArtifactChunk with the error field set. The orchestrator surfaces the error to the LLM.
Artifact exceeds 5 MB	The orchestrator rejects the artifact and returns an error to the LLM.
Artifact TTL expired (6 hours)	The orchestrator has evicted the stored artifact. The LLM receives an error when trying to reference the artifact_id.
Ownership mismatch (user_id/session_id)	The orchestrator refuses to upload the artifact to the target capability and returns an error.
UploadInputArtifact stream fails	The orchestrator reports a gRPC error to the LLM and the tool call is not executed.

Next steps

• Full proto reference — Complete capability.proto definition with all artifact message types (ArtifactChunk, DownloadOutputArtifactRequest, UploadInputArtifactChunk, UploadInputArtifactResponse).
• gRPC Interface — Overview of the full gRPC service contract.
• Container Guidelines — Network and resource rules for capability containers.