Metadata-Version: 2.4
Name: prismtrace-sdk
Version: 0.4.1
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"])
```

## LangChain Integration

Uses `X-PRISMtrace-Key`. Env vars `PRISMTRACE_API_KEY`, `PRISMTRACE_PROJECT_ID`,
and `PRISMTRACE_HOST` are read when constructor args are omitted.

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

handler = PRISMtraceCallbackHandler(
    api_key="pt-sk-...",
    project_id="your-project-id",
    host="https://api.prism.blockconvey.com",
    session_id="conversation-1",  # groups steps into a trajectory
)

llm = ChatAnthropic(model="claude-sonnet-4-5-20250514")
chain = LLMChain(llm=llm, prompt=prompt, callbacks=[handler])
result = chain.run("What is the credit risk for this customer?")
handler.flush()  # safe to call; also runs on process exit
```

## LangGraph Integration

```python
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)
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-2.0-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

```python
from prismtrace.otel 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 tracer.start_as_current_span("process_request") as span:
    span.set_attribute("prismtrace.span_type", "chain")
    span.set_attribute("prismtrace.input", "user query")
    # ... your code ...
    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=[...],
)
```
