Cookbook¶

Practical integration examples for every supported framework. Each section shows both the OSS path (offline, pip install pisama) and the Cloud path (platform API).

For full platform setup guides, see the dedicated Integration pages.

LangGraph¶

OSS: Analyze a LangGraph trace offline¶

Export your LangGraph execution as a trace dict and pass it to analyze():

import pisama
from langgraph.graph import StateGraph

# Run your LangGraph app
graph = StateGraph(...)
# ... define nodes and edges ...
app = graph.compile()
result = app.invoke({"input": "Research quantum computing trends"})

# Convert LangGraph state history to a Pisama trace
trace = {
    "trace_id": "lg-001",
    "spans": [
        {
            "name": node_name,
            "attributes": {
                "gen_ai.agent.name": node_name,
                "langgraph.node.name": node_name,
            },
            "input_data": step.get("input", {}),
            "output_data": step.get("output", {}),
        }
        for node_name, step in execution_history
    ],
}

analysis = pisama.analyze(trace)
for issue in analysis.issues:
    print(f"[{issue.type}] {issue.summary}")

Cloud: OTEL export¶

If your LangGraph app already exports OTEL traces, point the exporter to Pisama:

from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

exporter = OTLPSpanExporter(
    endpoint="https://your-pisama.com/api/v1/tenants/YOUR_TENANT_ID/traces/ingest",
    headers={"Authorization": "Bearer YOUR_TOKEN"},
)

For the full webhook-based setup, see LangGraph Integration.

CrewAI¶

OSS: Analyze CrewAI execution¶

import pisama
from crewai import Agent, Task, Crew

# Run your CrewAI pipeline
researcher = Agent(role="Researcher", goal="Find data", ...)
writer = Agent(role="Writer", goal="Write report", ...)
crew = Crew(agents=[researcher, writer], tasks=[...])
result = crew.kickoff()

# Build trace from CrewAI output
trace = {
    "trace_id": "crew-001",
    "spans": [
        {
            "name": task.description[:50],
            "attributes": {
                "gen_ai.agent.name": task.agent.role,
                "crewai.agent.role": task.agent.role,
            },
            "input_data": {"task": task.description},
            "output_data": {"result": task.output.raw if task.output else ""},
        }
        for task in crew.tasks
    ],
}

analysis = pisama.analyze(trace)
if analysis.has_issues:
    print(f"Found {len(analysis.issues)} issues in CrewAI execution")

Cloud: Webhook¶

Send CrewAI execution traces to the Pisama API after each run:

curl -X POST https://your-pisama.com/api/v1/tenants/YOUR_TENANT_ID/traces/ingest \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d @crewai_trace.json

AutoGen¶

OSS: Analyze AutoGen conversations¶

import pisama

# After an AutoGen group chat completes
chat_history = group_chat.messages  # list of message dicts

trace = {
    "trace_id": "autogen-001",
    "spans": [
        {
            "name": f"message_{i}",
            "attributes": {
                "gen_ai.agent.name": msg.get("name", msg.get("role", "unknown")),
                "autogen.agent.name": msg.get("name", "unknown"),
            },
            "input_data": {"role": msg.get("role", "")},
            "output_data": {"content": msg.get("content", "")},
        }
        for i, msg in enumerate(chat_history)
    ],
}

analysis = pisama.analyze(trace)
for issue in analysis.critical_issues:
    print(f"CRITICAL: [{issue.type}] {issue.summary}")

Claude Agent SDK¶

The pisama-agent-sdk package provides native hooks for the Claude Agent SDK.

pip install pisama-agent-sdk

Passive monitoring (hooks)¶

Attach hooks to capture every tool call:

from pisama_agent_sdk import pre_tool_use_hook, post_tool_use_hook

agent.hooks.pre_tool_use = pre_tool_use_hook
agent.hooks.post_tool_use = post_tool_use_hook

Pisama captures tool inputs/outputs and runs detection in the background. No changes to your agent logic.

Selective monitoring (matchers)¶

Monitor only specific tool categories:

from pisama_agent_sdk import (
    pre_tool_use_hook,
    post_tool_use_hook,
    FILE_TOOLS,
    SHELL_TOOLS,
    DANGEROUS_COMMANDS,
    create_matcher,
)

# Only monitor file and shell operations
matcher = create_matcher(FILE_TOOLS | SHELL_TOOLS)

Active self-check¶

Have the agent verify its own output before returning:

from pisama_agent_sdk import check

result = await check(
    output="The server is healthy based on the metrics.",
    context={"query": "Is auth-service down?", "sources": [...]},
)

if not result["passed"]:
    for issue in result["issues"]:
        print(f"Issue: {issue}")
    # Revise the output based on detected issues

Custom tool for Claude Agent SDK¶

Expose Pisama as a tool the agent can call:

from pisama_agent_sdk import create_check_tool
from claude_agent_sdk import ClaudeAgentOptions

options = ClaudeAgentOptions(
    custom_tools=[create_check_tool()],
)

The agent can then self-invoke Pisama checks during execution.

Evaluator mode¶

Use Pisama as an evaluator in multi-agent test harnesses:

from pisama_agent_sdk import PisamaEvaluator

evaluator = PisamaEvaluator()
eval_result = await evaluator.evaluate(trace_data)

for failure in eval_result.failures:
    print(f"{failure.type}: {failure.description}")

n8n¶

Webhook integration¶

Add an HTTP Request node at the end of your n8n workflow:

URL: https://your-pisama.com/api/v1/n8n/webhook
Method: POST
Headers:
  X-Pisama-API-Key: <your_api_key>
  Content-Type: application/json
Body: {{ $json }}

For the full setup guide including polling-based integration and n8n-specific detectors, see n8n Integration.

OSS: Analyze n8n execution logs offline¶

import pisama
import json

# Export execution data from n8n API or webhook capture
with open("n8n_execution.json") as f:
    execution = json.load(f)

trace = {
    "trace_id": execution["executionId"],
    "spans": [
        {
            "name": node["name"],
            "attributes": {"gen_ai.agent.name": node["name"]},
            "input_data": node.get("inputData", {}),
            "output_data": node.get("outputData", {}),
        }
        for node in execution.get("nodes", [])
    ],
}

analysis = pisama.analyze(trace)

Dify¶

Webhook integration¶

Configure your Dify app to POST execution data:

URL: https://your-pisama.com/api/v1/dify/webhook
Method: POST
Headers:
  X-Pisama-API-Key: <your_api_key>

For the full setup, see Dify Integration.

Generic OTEL¶

Any framework that exports OpenTelemetry traces can send them to Pisama.

Python (opentelemetry-sdk)¶

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

# Configure OTEL to export to Pisama
provider = TracerProvider()
exporter = OTLPSpanExporter(
    endpoint="https://your-pisama.com/api/v1/tenants/YOUR_TENANT_ID/traces/ingest",
    headers={"Authorization": "Bearer YOUR_TOKEN"},
)
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)

# Use gen_ai.* semantic conventions for best detection accuracy
tracer = trace.get_tracer("my-agent")
with tracer.start_as_current_span("agent_step") as span:
    span.set_attribute("gen_ai.agent.name", "my-agent")
    span.set_attribute("gen_ai.request.model", "claude-sonnet-4-20250514")
    span.set_attribute("gen_ai.usage.prompt_tokens", 1500)
    span.set_attribute("gen_ai.usage.completion_tokens", 800)
    # ... your agent logic ...

Key OTEL attributes¶

These gen_ai.* attributes improve detection accuracy:

Attribute	Purpose
`gen_ai.agent.name`	Agent identification
`gen_ai.request.model`	Model used for the step
`gen_ai.usage.prompt_tokens`	Input token count
`gen_ai.usage.completion_tokens`	Output token count
`gen_ai.state`	Agent state (for corruption detection)

Common patterns¶

CI/CD: Run detection in your test suite¶

import pytest
import pisama

def test_agent_no_loops():
    """Ensure the agent doesn't loop during execution."""
    result = run_my_agent()  # your agent execution
    trace = capture_trace(result)  # convert to trace dict

    analysis = pisama.analyze(trace)

    loop_issues = [i for i in analysis.issues if i.type == "loop"]
    assert len(loop_issues) == 0, f"Agent looped: {loop_issues[0].summary}"

def test_agent_no_hallucination():
    """Ensure the agent doesn't hallucinate."""
    result = run_my_agent()
    trace = capture_trace(result)

    analysis = pisama.analyze(trace)

    hallucination_issues = [i for i in analysis.issues if i.type == "hallucination"]
    assert len(hallucination_issues) == 0

Custom detector¶

from pisama_core import BaseDetector, DetectionResult, registry

class LatencyDetector(BaseDetector):
    name = "high_latency"

    async def detect(self, trace):
        slow_spans = [s for s in trace.spans if s.duration_ms and s.duration_ms > 30000]
        if slow_spans:
            return DetectionResult.issue_found(
                detector_name=self.name,
                summary=f"{len(slow_spans)} spans exceeded 30s",
                severity=50,
                confidence=0.95,
            )
        return DetectionResult.no_issue(self.name)

registry.register(LatencyDetector())

# Now analyze() will include your detector automatically
import pisama
result = pisama.analyze("trace.json")

Filtering by detector type¶

import pisama

result = pisama.analyze("trace.json")

# Only security-related issues
security = [i for i in result.issues if i.type in ("injection", "withholding")]

# Only critical issues (severity >= 60)
critical = result.critical_issues

# Check specific detector
has_loops = any(i.type == "loop" for i in result.issues)