Metadata-Version: 2.4
Name: prismtrace-sdk
Version: 0.4.3
Summary: PRISMtrace SDK — AI Observability by Block Convey
Author: Block Convey
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Requires-Dist: httpx>=0.24.0
Dynamic: author
Dynamic: description
Dynamic: description-content-type
Dynamic: requires-dist
Dynamic: requires-python
Dynamic: summary

# PRISMtrace Python SDK

AI Chat Observability by Block Convey. Supports Langchain, OpenTelemetry, Claude tool use tracing, **Trajectory evaluation**, and **Knowledge Base** management.

## Install

```bash
pip install prismtrace-sdk
```

Until the package is on PyPI, install from the repo (or set
`PRISMTRACE_SDK_INSTALL` in the PRISMtrace backend to the same command):

```bash
pip install "git+https://github.com/Block-Convey/prismtrace.git#subdirectory=sdk/python"
# or, from a local checkout:
pip install -e ./sdk/python
```

## Quickstart — Manual Trace

```python
from prismtrace import PRISMtrace

pt = PRISMtrace(
    api_key="pt-sk-your-key",
    host="https://api.prism.blockconvey.com",
    project_id="your-project-id",
)

pt.trace_llm(
    model="claude-sonnet-4-5-20250514",
    input_messages=[{"role": "user", "content": "Hello"}],
    output="Hi there!",
    latency_ms=320,
    token_count_input=10,
    token_count_output=5,
)

# Decorator
@pt.trace()
def ask_bot(question):
    # your LLM call here
    return "answer"
```

## Trajectory Evaluation

Submit agent trajectories (ordered step lists) for automated PRISM evaluation — goal adherence, tool compliance, efficiency, and safety scores.

```python
result = pt.submit_trajectory(
    agent_name="finance-agent",
    model="claude-sonnet-4-5-20250514",
    steps=[
        {
            "step_type": "reasoning",
            "label": "Analyze user query",
            "input_summary": "User asked about Q3 revenue",
            "output_summary": "Need to query knowledge base for financials",
            "duration_ms": 200,
            "token_count": 150,
        },
        {
            "step_type": "tool_call",
            "label": "knowledge_base_search",
            "tool_name": "knowledge_base_search",
            "input_summary": "Q3 revenue figures",
            "output_summary": "Found 3 matching documents",
            "duration_ms": 80,
        },
        {
            "step_type": "final_answer",
            "label": "Provide answer",
            "output_summary": "Q3 revenue was $4.2M, up 12% YoY",
            "duration_ms": 300,
            "token_count": 200,
        },
    ],
)

print(f"Trajectory ID: {result['id']}")

# Check evaluation results (runs async after submission)
evaluation = pt.get_trajectory_evaluation(result["id"])
print(evaluation)

# Re-trigger evaluation if config changed
pt.retrigger_evaluation(result["id"])
```

## Knowledge Base

Upload documents and search your project's knowledge base directly from the SDK.

```python
# Upload a document
doc = pt.kb_upload(
    filename="company-policy.md",
    content="# Return Policy\nAll items may be returned within 30 days...",
    description="Customer-facing return policy v2",
    content_type="text/markdown",
)
print(f"Document ID: {doc['id']}, Chunks: {doc['chunk_count']}")

# Search the knowledge base
results = pt.kb_search("return policy for electronics", limit=3)
for chunk in results:
    print(f"  Score: {chunk['score']:.2f} — {chunk['content'][:80]}...")

# List all documents
docs = pt.kb_list_documents()

# Delete a document
pt.kb_delete_document(doc["id"])
```

## Sessions

A trace is one run. A **session** groups the runs that belong together, and it
is what the platform analyses. Open one around a conversation:

```python
import prismtrace

with prismtrace.session(f"chat-{user_id}"):
    researcher.invoke(...)
    writer.invoke(...)     # same session, even sharing one handler
```

Build handlers once and share them. A `session_id=` passed to a constructor
pins that one value for the handler's whole life, so an app that builds one
handler per agent gets one session per agent. The session is read when a run
starts, so a handler built at import time still picks up the session open
around each call. Precedence is: an explicit `session_id` argument, then the
ambient session, then the handler's own id.

A context does not follow a plain thread or a process pool, so carry the id
across a queue or worker boundary by hand:

```python
session_id = prismtrace.current_session()      # producer
queue.put((job, session_id))

token = prismtrace.bind_session(session_id)    # worker
try:
    run(job)
finally:
    prismtrace.unbind_session(token)
```

Every handler holds an HTTP connection pool. Close it at shutdown, or use the
handler as a context manager:

```python
handler.close()          # flushes first; safe to call twice

with PRISMtraceCallbackHandler(api_key=..., project_id=...) as handler:
    ...
```

## LangChain Integration

Uses `X-PRISMtrace-Key`. Env vars `PRISMTRACE_API_KEY`, `PRISMTRACE_PROJECT_ID`,
and `PRISMTRACE_HOST` are read when constructor args are omitted. One handler is
safe to share across concurrent runs: each run gets its own trace.

```python
import prismtrace
from prismtrace import PRISMtraceCallbackHandler
from langchain.chains import LLMChain
from langchain_anthropic import ChatAnthropic

# Once, at startup.
handler = PRISMtraceCallbackHandler(
    api_key="pt-sk-...",
    project_id="your-project-id",
    host="https://api.prism.blockconvey.com",
)

llm = ChatAnthropic(model="claude-sonnet-4-5-20250514")
chain = LLMChain(llm=llm, prompt=prompt, callbacks=[handler])

with prismtrace.session("conversation-1"):
    result = chain.run("What is the credit risk for this customer?")

handler.close()  # at shutdown; flushes first, and flush also runs on exit
```

## LangGraph Integration

```python
import prismtrace
from prismtrace import PRISMtraceLangGraphHandler, wrap_langgraph

handler = PRISMtraceLangGraphHandler(
    api_key="pt-sk-...",
    project_id="your-project-id",
    host="https://api.prism.blockconvey.com",
    agent_name="support-graph",
)
graph = wrap_langgraph(compiled_graph, handler)

with prismtrace.session("conversation-1"):
    graph.invoke({"messages": [("user", "hello")]})
```

## Verify connection

```bash
export PRISMTRACE_HOST=https://your-host
export PRISMTRACE_PROJECT_ID=...
export PRISMTRACE_API_KEY=pt-sk-...
python -m prismtrace.verify
# Prints CREDENTIAL OK|FAIL and LIVE CONNECTED|WAITING FOR LIVE
```

## Google ADK Integration (Beta)

Capture Google Agent Development Kit workflows by passing the adapter's
hooks to an `LlmAgent`:

```python
from prismtrace import PRISMtraceADKAdapter
from google.adk.agents import LlmAgent

adapter = PRISMtraceADKAdapter(
    api_key="pt-sk-...",
    project_id="your-project-id",
    agent_name="my-adk-agent",
)

agent = LlmAgent(
    name="loan_assistant",
    model="gemini-3.6-flash",
    instruction="You help users understand loan products.",
    tools=[lookup_rate],
    before_model_callback=adapter.before_model,
    after_model_callback=adapter.after_model,
    before_tool_callback=adapter.before_tool,
    after_tool_callback=adapter.after_tool,
    before_agent_callback=adapter.before_agent,
    after_agent_callback=adapter.after_agent,
)
```

Wiring `before_tool_callback=adapter.before_tool` enables real tool
latency on `after_tool` traces; without it the trace stamps
`latency_ms=0` and `tool_latency_unavailable=True` in metadata.

### Error capture

ADK does not currently surface model/tool errors through a callback.
Emit explicit error traces from `try / except` blocks instead:

```python
try:
    response = await llm.generate_content_async(...)
except Exception as exc:
    adapter.record_model_error(exc, callback_context=ctx)
    raise

try:
    result = run_tool(args)
except Exception as exc:
    adapter.record_tool_error(exc, callback_context=ctx,
                              tool_name="lookup_rate",
                              tool=tool, tool_context=ctx)
    raise

try:
    await Runner.run_async(...)
except Exception as exc:
    adapter.record_runner_error(exc)
    raise
```

Each emits a trace with `status=error` in metadata. Error messages are
scrubbed for `pt-sk-…`, `sk-…`, `sk-ant-…`, `AIza…` and other known
secret patterns before they're posted.

Runnable demo: [`examples/google_adk_demo.py`](examples/google_adk_demo.py).
Full validation notes (what's captured, Beta gaps, known issues):
[`docs/GOOGLE_ADK_INTEGRATION.md`](../../docs/GOOGLE_ADK_INTEGRATION.md).

## OpenTelemetry Integration

There are two ways to get OpenTelemetry data into PRISM, and for most people
this SDK exporter is **not** the one to reach for.

`POST /api/otlp/v1/traces` accepts OTLP over HTTP directly, protobuf or JSON
encoded, so an instrumented service in any language can point its existing
exporter at PRISM with no SDK at all. One limit applies: gRPC is not served, so
an exporter that defaults to it (the Java SDK does) needs
`OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf`. See ADR-0141 and ADR-0148.

`PRISMtraceInstrumentor` below is the in-process alternative: it converts OTel
spans to PRISM's native span ingest without leaving your application. Prefer it
only when you are already using this SDK and do not want to run an exporter or
a Collector.

```python
import prismtrace
from prismtrace import PRISMtraceInstrumentor

instrumentor = PRISMtraceInstrumentor()
instrumentor.instrument(
    api_key="pt-sk-...",
    project_id="your-project-id",
    endpoint="https://api.prism.blockconvey.com",
)

tracer = instrumentor.get_tracer("my-service")
with prismtrace.session("conversation-1"):
    with tracer.start_as_current_span("process_request"):
        ...  # your code

instrumentor.shutdown()   # flushes pending spans and closes the pool
```

If you are already instrumented with a GenAI library, no extra attributes are
needed. The exporter reads the OpenTelemetry GenAI semantic conventions:

| Field | Attributes read, in order |
| --- | --- |
| model | `gen_ai.response.model`, `gen_ai.request.model` |
| input tokens | `gen_ai.usage.input_tokens`, `gen_ai.usage.prompt_tokens` |
| output tokens | `gen_ai.usage.output_tokens`, `gen_ai.usage.completion_tokens` |
| input text | `gen_ai.input.messages`, `gen_ai.prompt` |
| output text | `gen_ai.output.messages`, `gen_ai.completion` |
| span type | `gen_ai.operation.name` |
| session | `gen_ai.conversation.id`, `session.id` |

The second spelling in each row is the superseded one, kept because
instrumentation in the wild still emits it. Setting a `prismtrace.*` attribute
overrides whatever the conventions say:

```python
with tracer.start_as_current_span("process_request") as span:
    span.set_attribute("prismtrace.span_type", "chain")
    span.set_attribute("prismtrace.input", "user query")
    span.set_attribute("prismtrace.output", "response")
```

Requires: `pip install opentelemetry-sdk opentelemetry-api`

## Claude Tool Use Tracing (with auto-trajectory)

The `ClaudeAgentTracer` now automatically emits both **spans** (for trace detail) and a **trajectory** (for PRISM evaluation) after each agentic run.

```python
import anthropic
from prismtrace.claude_tracer import ClaudeAgentTracer

client = anthropic.Anthropic()

tracer = ClaudeAgentTracer(
    anthropic_client=client,
    api_key="pt-sk-...",
    project_id="your-project-id",
    endpoint="https://api.prism.blockconvey.com",
    agent_name="weather-agent",    # shows up in trajectory analytics
    emit_trajectory=True,          # default: auto-submit trajectory
)

tools = [
    {
        "name": "get_weather",
        "description": "Get the weather for a location",
        "input_schema": {
            "type": "object",
            "properties": {"location": {"type": "string"}},
            "required": ["location"],
        },
    }
]

def execute_tool(name: str, input_data: dict) -> str:
    if name == "get_weather":
        return f"72F and sunny in {input_data['location']}"
    return "Unknown tool"

result = tracer.run(
    messages=[{"role": "user", "content": "What's the weather in SF?"}],
    tools=tools,
    system="You are a helpful assistant.",
    tool_executor=execute_tool,
)

print(f"Trace ID:       {result['trace_id']}")
print(f"Trajectory ID:  {result['trajectory_id']}")
print(f"Iterations:     {result['iterations']}")
print(f"Traj. steps:    {result['trajectory_steps']}")
```

### Auto-instrument all Claude calls

```python
tracer.instrument_client()

# Now every client.messages.create() call is automatically traced
response = client.messages.create(
    model="claude-sonnet-4-5-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
)
```

## Zero-code Proxy

If you route LLM calls through the PRISMtrace proxy (`/proxy/anthropic/v1/messages`), trajectories are **automatically created** whenever the response contains tool-use blocks. No SDK changes needed — just point your Anthropic base URL at the proxy.

```python
import anthropic

client = anthropic.Anthropic(
    base_url="https://api.prism.blockconvey.com/proxy/anthropic/v1",
    default_headers={"X-PRISMtrace-Key": "pt-sk-your-key"},
)

# Every tool-use response automatically gets a trajectory + PRISM evaluation
response = client.messages.create(
    model="claude-sonnet-4-5-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "What's the weather?"}],
    tools=[...],
)
```
