Module: artifacts¶
from adk_fluent import A
Artifact operations — bridge between state and artifact service.
Quick Reference¶
Method |
Returns |
Description |
|---|---|---|
|
|
Publish state content to artifact service |
|
|
Snapshot artifact content into state |
|
|
Watch an artifact: load its content into state[into] for observation |
|
|
Record an artifact’s version metadata into state for change detection |
|
|
Bridge an artifact write into a state signal, then run |
|
|
Save literal content to artifact service. |
|
|
Load artifact for pipeline composition. |
|
|
List artifact filenames into state. |
|
|
Get artifact version metadata into state. |
|
|
Delete all versions of an artifact. |
|
|
Parse JSON string in state[key] to dict/list |
|
|
Parse CSV string in state[key] to list[dict] |
|
|
Ensure state[key] is a decoded string. |
|
|
Serialize state[key] dict/list to JSON string |
|
|
Serialize state[key] list[dict] to CSV string |
|
|
Convert Markdown state[key] to HTML string |
|
|
Conditional artifact operation. |
|
|
Load artifact directly into LLM context. |
|
|
Batch publish: multiple (filename, from_key) pairs |
|
|
Batch snapshot: multiple (filename, into_key) pairs |
Methods¶
A.publish(filename: str, *, from_key: str, mime: str | None = None, metadata: dict[str, Any] | None = None, scope: "Literal[session, user]" = session) -> ATransform¶
Publish state content to artifact service.
STATE BRIDGE: reads state[from_key], copies to versioned artifact.
Parameters:
filename(str)from_key(str)mime(str | None) — default:Nonemetadata(dict[str, Any] | None) — default:Nonescope(Literal[‘session’, ‘user’]) — default:'session'
A.snapshot(filename: str, *, into_key: str, version: int | None = None, decode: bool = False, scope: "Literal[session, user]" = session) -> ATransform¶
Snapshot artifact content into state.
STATE BRIDGE: loads artifact, copies point-in-time content into state[into_key].
Parameters:
filename(str)into_key(str)version(int | None) — default:Nonedecode(bool) — default:Falsescope(Literal[‘session’, ‘user’]) — default:'session'
A.watch(filename: str, *, into: str, version: int | None = None, decode: bool = False, scope: "Literal[session, user]" = session) -> ATransform¶
Watch an artifact: load its content into state[into] for observation.
SUBSCRIBE / OBSERVATION DUAL of A.publish. Where publish copies state -> artifact, watch copies artifact -> state so downstream steps and R reactor rules can observe artifact changes by reading state[into].
Semantics are intentionally identical to A.snapshot at runtime
(artifact -> state, latest version unless version is pinned) so
it reuses the existing, battle-tested artifact bridge with no new
runtime op. The distinction is one of intent:
A.snapshotis a one-shot point-in-time copy into a result key.A.watchis meant to be re-run (inside a Loop, or each turn of a change-detection pipeline) to keep state[into] tracking the latest artifact content. Pair withA.watch_versionorA.on_changeto make a change detectable by an R signal.
Example (change-detection loop): ( A.watch(“inbox.json”, into=”inbox”) >> A.watch_version(“inbox.json”, into=”inbox_version”) >> processor ) * until(lambda s: s.get(“done”))
Args:
filename: Artifact to load (latest version by default).into: State key to write the artifact content into.version: Pin a specific version;Nonemeans latest.decode: Decode binary inline_data as UTF-8 text (mirrors snapshot).scope:"session"(default) or"user".
Parameters:
filename(str)into(str)version(int | None) — default:Nonedecode(bool) — default:Falsescope(Literal[‘session’, ‘user’]) — default:'session'
A.watch_version(filename: str, *, into: str, scope: "Literal[session, user]" = session) -> ATransform¶
Record an artifact’s version metadata into state for change detection.
Loads the latest version metadata of filename into state[into]
as a dict {"version", "mime_type", "create_time", "canonical_uri"}.
Because the artifact service bumps version on every write, this is
the cheap, content-free signal an R rule can fire on:
ver = R.signal(“inbox_version”)
handler = agent.on(R.changed(“inbox_version”))
pipeline = A.watch_version("inbox.json", into="inbox_version") >> ...
SUBSCRIBE DUAL: pairs with A.watch (content) — watch_version is the
lightweight change-trigger, watch is the content load. Reuses the
existing version artifact op, so there is no new runtime path.
Args:
filename: Artifact whose version to record.into: State key to write the version metadata dict into.scope:"session"(default) or"user".
Parameters:
filename(str)into(str)scope(Literal[‘session’, ‘user’]) — default:'session'
A.on_change(filename: str, handler: Any, *, into: str | None = None, version_key: str | None = None, scope: "Literal[session, user]" = session) -> tuple[ATransform, ...]¶
Bridge an artifact write into a state signal, then run handler.
Returns a tuple of pipeline steps that, when run, (1) record the
artifact’s version into state[version_key] so a change is
detectable, (2) load its content into state[into] for the handler
to consume, and (3) run handler (any builder / agent / function).
Unpack into a pipeline like A.publish_many (the tuple is not
>>-chainable directly — feed the steps to Pipeline.step):
steps = A.on_change(“inbox.json”, processor, into=”inbox”)
pipeline = Pipeline(“react”).step(ingest)
for step in steps:
pipeline = pipeline.step(step)
This is the artifact analogue of builder.on(R.changed(...)): it
gives you a guaranteed, reactor-free “fire on artifact write” step.
For true asynchronous reactor firing, register an R rule on
version_key and feed it A.watch_version (see that method);
on_change is the synchronous, in-pipeline form.
Args:
filename: Artifact to watch.handler: Builder/agent/callable to run after loading the artifact.into: State key for the loaded content. Defaults to a derived key.version_key: State key for the version signal. Defaults to"<into>_version".scope:"session"(default) or"user".
Returns:
`(watch_version_step, watch_step, handler)` — chain with `>>`.
Parameters:
filename(str)handler(Any)into(str | None) — default:Noneversion_key(str | None) — default:Nonescope(Literal[‘session’, ‘user’]) — default:'session'
A.save(filename: str, *, content: str | bytes, mime: str | None = None, metadata: dict[str, Any] | None = None, scope: "Literal[session, user]" = session) -> ATransform¶
Save literal content to artifact service. No state bridge.
Parameters:
filename(str)content(str | bytes)mime(str | None) — default:Nonemetadata(dict[str, Any] | None) — default:Nonescope(Literal[‘session’, ‘user’]) — default:'session'
A.load(filename: str, *, scope: "Literal[session, user]" = session) -> ATransform¶
Load artifact for pipeline composition. No state bridge.
Parameters:
filename(str)scope(Literal[‘session’, ‘user’]) — default:'session'
A.list(*, into_key: str) -> ATransform¶
List artifact filenames into state. Lightweight metadata only.
Parameters:
into_key(str)
A.version(filename: str, *, into_key: str) -> ATransform¶
Get artifact version metadata into state. Lightweight metadata only.
Parameters:
filename(str)into_key(str)
A.delete(filename: str) -> ATransform¶
Delete all versions of an artifact. No state involvement.
Parameters:
filename(str)
A.as_json(key: str) -> STransform¶
Parse JSON string in state[key] to dict/list.
Usage: A.snapshot(“data.json”, into_key=”data”) >> A.as_json(“data”)
Parameters:
key(str)
A.as_csv(key: str, *, columns: list[str] | None = None) -> STransform¶
Parse CSV string in state[key] to list[dict].
Usage: A.snapshot(“data.csv”, into_key=”rows”) >> A.as_csv(“rows”)
Parameters:
key(str)columns(list[str] | None) — default:None
A.as_text(key: str, *, encoding: str = utf-8) -> STransform¶
Ensure state[key] is a decoded string. Decodes bytes if needed.
Usage: A.snapshot(“raw.bin”, into_key=”text”) >> A.as_text(“text”)
Parameters:
key(str)encoding(str) — default:'utf-8'
A.from_json(key: str, *, indent: int | None = None) -> STransform¶
Serialize state[key] dict/list to JSON string.
Usage: A.from_json(“config”) >> A.publish(“config.json”, from_key=”config”)
Parameters:
key(str)indent(int | None) — default:None
A.from_csv(key: str) -> STransform¶
Serialize state[key] list[dict] to CSV string.
Usage: A.from_csv(“rows”) >> A.publish(“results.csv”, from_key=”rows”)
Parameters:
key(str)
A.from_markdown(key: str) -> STransform¶
Convert Markdown state[key] to HTML string.
Uses Python’s built-in markdown if available, falls back to minimal conversion. Usage: A.from_markdown(“report”) >> A.publish(“report.html”, from_key=”report”)
Parameters:
key(str)
A.when(predicate: str | Callable, transform: ATransform) -> ATransform¶
Conditional artifact operation. Uniform with S.when(), C.when(), etc.
Parameters:
predicate(str | Callable)transform(ATransform)
A.for_llm(filename: str, *, version: int | None = None, scope: "Literal[session, user]" = session) -> CTransform¶
Load artifact directly into LLM context. No state bridge.
Text artifacts are decoded and injected as instruction context. Binary artifacts get a placeholder description with MIME and size. Composes with C module: Agent(“x”).context(C.from_state(“topic”) + A.for_llm(“report.md”))
Parameters:
filename(str)version(int | None) — default:Nonescope(Literal[‘session’, ‘user’]) — default:'session'
A.publish_many(*pairs: tuple[str, str], mime: str | None = None, scope: "Literal[session, user]" = session) -> tuple[ATransform, ...]¶
Batch publish: multiple (filename, from_key) pairs.
Usage: Agent(“w”).artifacts(*A.publish_many((“r.md”, “report”), (“d.json”, “data”)))
Parameters:
*pairs(tuple[str, str])mime(str | None) — default:Nonescope(Literal[‘session’, ‘user’]) — default:'session'
A.snapshot_many(*pairs: tuple[str, str], scope: "Literal[session, user]" = session) -> tuple[ATransform, ...]¶
Batch snapshot: multiple (filename, into_key) pairs.
Usage: Agent(“r”).artifacts(*A.snapshot_many((“r.md”, “text”), (“d.json”, “data”)))
Parameters:
*pairs(tuple[str, str])scope(Literal[‘session’, ‘user’]) — default:'session'
Types¶
Type |
Description |
|---|---|
|
Composable artifact operation descriptor |