Spineforge SDK

AI agent identity and observability — Okta + Datadog for the AI agent era.

Homepage | Documentation

Spineforge assigns every AI agent a stable identity (Spine ID), instruments LLM and tool calls via OpenLLMetry (OpenTelemetry-based auto-instrumentation), and groups events into Runs and Actions that map directly to the hosted Spineforge backend.

---

Quick Start

Install

pip install spineforge

# For running the example agents
pip install "spineforge[examples]"

Usage (3 lines to integrate)

import spineforge

# 1. Init — resolves/creates a stable Spine ID for this agent
spine = spineforge.init(agent_name="my-research-agent")

# 2. Wrap your agent's top-level invocation in a run
with spine.run(input=user_query) as run:
    result = run_agent(user_query)
    run.set_output(result)

That's it. Spineforge auto-instruments LLM calls (Groq, LangChain) in the background via OpenTelemetry. No code changes to your agent logic.

For tool functions not auto-captured

@spineforge.track_tool
def call_tool(name: str, query: str) -> str:
    """Only needed for raw SDK agents — LangChain tools are auto-captured."""
    return tools[name].run(query)

---

How It Works

init(agent_name) → Spine

1. Resolves identity — looks up agent_name in .spineforge/registry.json. If it exists, returns the same Spine ID. If not, generates a new UUID-4 and persists it.
2. Wires instrumentation — creates an OpenTelemetry TracerProvider, registers a custom SpineforgeSpanProcessor, and activates OpenLLMetry instrumentors for installed libraries (Groq, LangChain).
3. Returns a Spine handle — use spine.run() to group actions into runs.

spine.run(input=...) → context manager

The with spine.run(...) block:

• Generates a run_id and sets it in a ContextVar so all spans created during the block are associated with this run.
• Emits a RunEvent(status="running") at entry.
• Emits a RunEvent(status="success") or RunEvent(status="error") at exit.
• Force-flushes the OTel pipeline to ensure all action spans are written before the run-end event.

Span → ActionEvent translation

The SpineforgeSpanProcessor (a custom OTel SpanProcessor) receives every completed span and classifies it:

Source	Span attribute	ActionEvent type
Groq instrumentor	gen_ai.system present	llm_call
LangChain instrumentor	traceloop.span.kind = "llm"	llm_call
LangChain instrumentor	traceloop.span.kind = "tool"	tool_call
@track_tool decorator	spineforge.tool_call = True	tool_call

Unrecognised spans (HTTP, framework internals) are silently ignored.

---

Sink Abstraction

All events flow through a Sink interface:

class Sink(ABC):
    def emit(self, event: dict) -> None: ...
    def flush(self) -> None: ...
    def shutdown(self) -> None: ...

Built-in sinks:

• ConsoleSink — compact, coloured one-liners to stdout
• FileSink — non-blocking JSONL append via background thread + queue
• APISink — batched, non-blocking POSTs to the Spineforge backend with automatic retry and local file fallback

Adding a custom sink (e.g. SupabaseSink):

from spineforge.sinks import Sink

class SupabaseSink(Sink):
    def __init__(self, supabase_url: str, api_key: str):
        self._client = create_supabase_client(supabase_url, api_key)

    def emit(self, event: dict) -> None:
        table = "runs" if event["event"] == "run" else "actions"
        self._client.table(table).insert(event).execute()

    def flush(self) -> None:
        pass  # or batch-flush pending writes

    def shutdown(self) -> None:
        self._client.close()

No changes to instrumentation.py or span_processor.py needed — just add the sink to the list in __init__.py:init().

---

Adding a New Framework Instrumentor

To add support for a new framework (e.g. CrewAI):

1. Install the instrumentor package

pip install opentelemetry-instrumentation-crewai

2. Add activation in instrumentation.py

# In _instrument_libraries():
try:
    from opentelemetry.instrumentation.crewai import CrewAIInstrumentor
    CrewAIInstrumentor().instrument(tracer_provider=provider)
    _log_instrumentor_status("CrewAI", True)
except ImportError:
    _log_instrumentor_status("CrewAI", False, "not installed")

3. Add span classification in span_processor.py

Add a new condition in _classify_and_build() to detect CrewAI-specific span attributes and route them to _build_llm_event() or a new tool event builder.

4. Add the dependency to pyproject.toml

"opentelemetry-instrumentation-crewai>=0.35.0",

---

Which Example Agents Need @track_tool?

Agent	File	@track_tool needed?	Why
Scratch (raw Groq)	scratch_agent_demo.py	Yes	The Groq instrumentor only wraps client.chat.completions.create() — it has no knowledge of tool execution. call_tool() must be decorated.
LangChain	langchain_agent_demo.py	No	The LangChain instrumentor hooks into LangChain's callback system (on_tool_start/on_tool_end), which fires automatically when Tool.run() is called via AgentExecutor.

---

Output Formats

Console (human-readable)

🦴 Spineforge initialised: agent='research-agent-scratch'  spine_id=a1b2c3d4…
🦴 [SPINE:a1b2c3d4] RUN started  input="What is quantum computing?"
🦴 [SPINE:a1b2c3d4] [RUN:e5f6g7h8] ⚡ llm_call  llama-3.1-8b-instant  342ms ✓  (prompt:128 comp:256)
🦴 [SPINE:a1b2c3d4] [RUN:e5f6g7h8] 🔧 tool_call  search_tool  1204ms ✓
🦴 [SPINE:a1b2c3d4] [RUN:e5f6g7h8] ⚡ llm_call  llama-3.1-8b-instant  567ms ✓  (prompt:384 comp:512)
🦴 [SPINE:a1b2c3d4] [RUN:e5f6g7h8] RUN completed  4521ms ✓

JSONL (.spineforge/logs/actions.jsonl)

Each line is a JSON object — either a run event or an action event:

{"event": "run", "run_id": "...", "spine_id": "...", "agent_name": "...", "status": "running", ...}
{"event": "action", "action_id": "...", "run_id": "...", "type": "llm_call", "name": "llama-3.1-8b-instant", ...}
{"event": "action", "action_id": "...", "run_id": "...", "type": "tool_call", "name": "search_tool", ...}
{"event": "run", "run_id": "...", "status": "success", "ended_at": "...", ...}

---

Configuration

Env var	Default	Description
SPINEFORGE_DATA_DIR	CWD	Base directory for .spineforge/
SPINEFORGE_LOG_FILE	actions.jsonl	JSONL log filename
SPINEFORGE_VERBOSE	true	Enable console output
SPINEFORGE_TRACE_CONTENT	true	Capture prompt/completion text
SPINEFORGE_REGISTRY_URL	"" (empty)	URL of the Spineforge backend. Empty means offline mode.

All can also be passed as kwargs to spineforge.init().

---

Project Structure

spineforge-sdk/
├── pyproject.toml              # Package metadata + dependencies
├── README.md                   # This file
├── spineforge/
│   ├── __init__.py             # Public API: init(), track_tool, Spine, Run
│   ├── config.py               # SpineforgeConfig (env-configurable)
│   ├── models.py               # RunEvent, ActionEvent dataclasses
│   ├── identity.py             # Spine ID registry (registry.json)
│   ├── sinks.py                # Sink ABC + ConsoleSink + FileSink
│   ├── span_processor.py       # SpineforgeSpanProcessor (OTel → events)
│   ├── instrumentation.py      # TracerProvider + instrumentor wiring
│   └── decorators.py           # @track_tool
├── examples/
│   ├── scratch_agent_demo.py   # scratch.py + spineforge
│   ├── langchain_agent_demo.py # main.py + spineforge
│   └── tools.py                # Shared tools
└── .spineforge/                # Created at runtime (gitignored)
    ├── registry.json
    └── logs/
        └── actions.jsonl

---

License

This software is proprietary and confidential. Unauthorized copying, distribution, reproduction, publication, or modification of this software, via any medium, is strictly prohibited. For internal use only.