Configuration

Sinks

Configure where Spineforge sends telemetry events — console, file, backend, or your own.


Environment variables

VariableDefaultDescription
SPINEFORGE_DATA_DIRCWDBase directory for .spineforge/ (registry + logs)
SPINEFORGE_LOG_FILEactions.jsonlJSONL log filename inside .spineforge/logs/
SPINEFORGE_VERBOSEtrueEnable ConsoleSink (coloured one-liners to stdout)
SPINEFORGE_TRACE_CONTENTtrueCapture prompt / completion text in action events
SPINEFORGE_REGISTRY_URL(empty)Backend URL — empty = offline mode (no APISink)

All can also be passed as kwargs to spineforge.init() — kwargs win over env vars.

ConsoleSink

Prints a compact, coloured one-liner per event to stdout. Enabled when SPINEFORGE_VERBOSE=true.

🦴 [SPINE:a1b2c3d4] RUN started  input="What is the capital of France?"
🦴 [SPINE:a1b2c3d4] [RUN:e5f6g7h8] ⚡ llm_call  gpt-4o  612ms ✓  (prompt:48 comp:12)
🦴 [SPINE:a1b2c3d4] [RUN:e5f6g7h8] 🔧 tool_call  search_web  341ms ✓
🦴 [SPINE:a1b2c3d4] [RUN:e5f6g7h8] RUN completed  1024ms ✓

To disable: SPINEFORGE_VERBOSE=false or spineforge.init(..., verbose=False)

FileSink

Appends events as JSON lines to <data_dir>/.spineforge/logs/actions.jsonl. Non-blocking — events are pushed to an in-memory queue and written by a background daemon thread.flush() drains the queue synchronously (called automatically at process exit via atexit).

FileSink is always active. It also serves as the fallback for APISink on connection failures.

APISink

Active when SPINEFORGE_REGISTRY_URL is set. POSTs telemetry batches to the Spineforge backend at /telemetry/ingest. Authenticated with a scoped JWT (telemetry:write scope). Batch size: up to 50 events, flushed every 5 seconds.

  • 200 — all events accepted
  • 207 — partial success; failed events fall back to FileSink
  • Non-2xx / connection error — entire batch falls back to FileSink

Custom sinks

Implement the Sink ABC and pass it to init():

from spineforge.sinks import Sink

class MyDataWarehouseSink(Sink):
    def emit(self, event: dict) -> None:
        if event["event"] == "action":
            self.warehouse.insert("agent_actions", event)

    def flush(self) -> None:
        self.warehouse.commit()

    def shutdown(self) -> None:
        self.flush()
        self.warehouse.close()

No changes to instrumentation or span processing needed — just add the sink to your init call.

Custom sink support via init(extra_sinks=[...]) is planned. Currently, custom sinks require a light wrapper around spineforge.init().

Disabling content capture

If you don't want prompt or completion text stored (e.g. for compliance):

spine = spineforge.init(
    agent_name="my-agent",
    trace_content=False   # or SPINEFORGE_TRACE_CONTENT=false
)

Token counts and costs are still captured. Only the text content is omitted.

Next steps

  • SSO / OIDC — configure agent ownership via your IdP
  • Telemetry — how the sink pipeline connects to the OTel span processor