Save and Load Artifacts

Kitaru offers two ways to work with artifacts, depending on where your code runs.

Inside checkpoints

Use kitaru.save() and kitaru.load() for inline artifact handling:

• kitaru.save(name, value, ...) stores a named value in the current checkpoint.
• kitaru.load(exec_id, name) reads a named value from another execution's checkpoint.

Both are checkpoint-only. Calling them outside @checkpoint raises an error.

Outside checkpoints

Use the client APIs for browsing artifacts from completed executions:

• KitaruClient().executions.get(exec_id).artifacts — browse all artifacts as ArtifactRef objects.
• KitaruClient().artifacts.get(artifact_id) — get a reference to a specific artifact.

Call .load() on any ArtifactRef to materialize the actual value:

artifact = KitaruClient().artifacts.get(artifact_id)
value = artifact.load()

Artifact types

Kitaru validates these artifact types:

• prompt
• response
• context
• input
• output
• blob

Example

from kitaru import checkpoint
import kitaru

@checkpoint
def research(topic: str) -> str:
    notes = f"Research notes about {topic}."
    kitaru.save("research_context", {"topic": topic, "notes": notes}, type="context")
    return notes

@checkpoint
def follow_up(prev_exec_id: str) -> str:
    # Load checkpoint output by checkpoint name.
    previous_notes = kitaru.load(prev_exec_id, "research")
    # Load explicit named artifact.
    context = kitaru.load(prev_exec_id, "research_context")
    return f"{previous_notes} [topic={context['topic']}]"

Name lookup behavior

When loading from an execution:

• checkpoint outputs are addressable by checkpoint name
• manual saved artifacts are addressable by artifact name

If no match is found, kitaru.load() raises an error. If multiple matches are found, kitaru.load() raises an ambiguity error so you can choose a unique name.

For custom invocation IDs or heavily dynamic checkpoint naming, prefer explicit kitaru.save() names for stable lookup.

Behavior for artifacts from failed or incomplete source executions follows what is persisted in runtime storage and may evolve in later releases.

Try it locally

uv sync --extra local
uv run examples/basic_flow/flow_with_artifacts.py
uv run pytest tests/test_phase8_artifacts_example.py

For the broader catalog, see Examples.