invocation.md

Invocation

Etch handlers can be called in streaming mode (yield events as they happen) or non-streaming mode (drain the full run into a single result object).

Streaming: etch.stream()

for event in etch.stream("investigate", address="bc1q..."):
    if event.kind == "reasoning":
        print(event.payload)
    elif event.kind == "tool_call":
        print(f"Calling {event.payload['tool']}")
    elif event.is_terminal():
        print("Done:", event.payload)

Returns an iterator of EtchEvent. Events are normalized — handlers can yield tuples or EtchEvent directly.

Optional session_id overrides the auto-generated session identifier.

Non-streaming: etch.run()

result = etch.run("investigate", address="bc1q...")
print(result.ok)
print(result.result)       # terminal payload
print(result.trace)        # list of progress events
print(result.tool_call_count)
print(result.duration_ms)
print(result.metrics)    # token stats (see Metrics guide)

Returns an EtchRunResult.

EtchRunResult fields

Field	Type	Description
ok	bool	False when terminal payload has "ok": False
handler	str	Handler name invoked
session_id	str	Unique run identifier
terminal_kind	str | None	e.g. "investigate_result"
result	Any	Terminal event payload
trace	list[dict]	Progress events {kind, payload}
tool_call_count	int	Non-cached tool invocations
duration_ms	float	Wall-clock run time
metrics	dict | None	Token breakdown
report	str | None	Markdown from report_builder
error	str | None	From terminal "error" key

Serialize for APIs:

result.to_dict()

Callable shorthand: etch()

result = etch("investigate", address="...")              # same as run()
stream = etch("investigate", address="...", stream=True)  # same as stream()

Lower-level utilities

collect_run()

Drain an arbitrary event iterable without an Etch instance:

from etch_sdk import collect_run

result = collect_run(
    my_generator(),
    handler="custom",
    session_id="sess-1",
    tool_call_count=3,
    report_builder=lambda t: render_report(t),
)

drain()

Call a stream function and collect:

from etch_sdk import drain, EtchContext

ctx = EtchContext(session_id="s1", handler="analyze")
result = drain(analyze_fn, "analyze", session_id="s1", ctx=ctx, text="hello")

Useful when integrating with existing generator functions outside the Etch class.

Choosing a mode

Use streaming when	Use non-streaming when
Building a live UI	Returning JSON from an HTTP endpoint
Showing reasoning as it happens	Caching full run results
User may cancel mid-run	You need trace + metrics in one object
SSE / WebSocket transport	Batch or CLI processing

The same handler implementation supports both — no duplication.

Report and structured builders

Pass callbacks to run() to derive extra fields:

result = etch.run(
    "investigate",
    address=addr,
    report_builder=lambda t: generate_markdown(t),
    structured_builder=lambda t: {"paths": t["path_results"]},
)

• report on the result holds the markdown string
• structured_builder output feeds metrics token counting for the structured portion

Related

• Events — what streaming yields
• Streaming & HTTP — SSE wrapper around stream()
• Metrics — how result.metrics is computed