SwisperStudio SDK Integration Guide¶

# Install from PyPI (public package)
pip install swisper-studio-sdk==0.5.2

# Verify it works
python -c "from swisper_studio_sdk import create_traced_graph; print('✅ Ready!')"

# Check Python version (3.11+)
python --version

# Check pip is installed
pip --version

# Check Redis is running
redis-cli ping
# Should return: PONG

cd /path/to/swisper_studio
docker compose up -d

pip install swisper-studio-sdk==0.5.2

# Test import
python -c "from swisper_studio_sdk import create_traced_graph, initialize_redis_publisher; print('✅ SDK installed successfully!')"

# Check installed version
pip show swisper-studio-sdk

# Expected output:
# Name: swisper-studio-sdk
# Version: 0.5.2
# Summary: SwisperStudio SDK - Tracing integration for Swisper (internal use)
# Location: /path/to/site-packages

# requirements.txt

# SwisperStudio SDK from PyPI
swisper-studio-sdk==0.5.2

pip install -r requirements.txt

# Install dependencies from requirements.txt
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

swisper-studio-sdk==0.5.2

docker compose build backend
docker compose up -d backend

# Test import
python -c "from swisper_studio_sdk import create_traced_graph, initialize_redis_publisher; print('✅ SDK installed successfully!')"

# Check version
python -c "import swisper_studio_sdk; print(f'SDK version: 0.5.2')"

✅ SDK installed successfully!

# SwisperStudio Integration (Observability & Configuration Management)
SWISPER_STUDIO_ENABLED: bool = True
SWISPER_STUDIO_REDIS_URL: str = "redis://redis:6379"  # Redis for observability events
SWISPER_STUDIO_STREAM_NAME: str = "observability:events"
SWISPER_STUDIO_PROJECT_ID: str = "0d7aa606-cb29-4a31-8a59-50fa61151a32"  # Your project ID
SWISPER_STUDIO_CAPTURE_REASONING: bool = True  # Capture LLM reasoning (<think> tags)
SWISPER_STUDIO_REASONING_MAX_LENGTH: int = 50000  # 50 KB limit

cd backend
python -c "from app.core.config import settings; print(f'Enabled: {settings.SWISPER_STUDIO_ENABLED}')"
# Should print: Enabled: True

# Initialize SwisperStudio observability (SDK v0.5.2)
if settings.SWISPER_STUDIO_ENABLED:
    try:
        from swisper_studio_sdk import initialize_redis_publisher, wrap_llm_adapter

        # Initialize Redis Streams publisher for observability
        await initialize_redis_publisher(
            redis_url=settings.SWISPER_STUDIO_REDIS_URL,
            stream_name=settings.SWISPER_STUDIO_STREAM_NAME,
            project_id=settings.SWISPER_STUDIO_PROJECT_ID,
            verify_consumer=True,  # Check if SwisperStudio consumer is running
        )
        logger.info("✅ SwisperStudio observability initialized (Redis Streams)")
        logger.info(f"   Redis: {settings.SWISPER_STUDIO_REDIS_URL}")
        logger.info(f"   Stream: {settings.SWISPER_STUDIO_STREAM_NAME}")
        logger.info(f"   Project ID: {settings.SWISPER_STUDIO_PROJECT_ID}")

        # ⚠️ CRITICAL: Enable LLM prompt capture for AUTO type detection
        # Without this, ALL observations will be SPAN (not GENERATION)!
        try:
            wrap_llm_adapter()
            logger.info("✅ LLM prompt capture enabled (structured + streaming)")
            logger.info("   AUTO type detection will work for @traced decorators")
        except Exception as e:
            logger.warning(f"⚠️ LLM prompt capture disabled: {e}")
            logger.warning("   LLM nodes will show as SPAN instead of GENERATION!")
            logger.warning("   Workaround: Use @traced(observation_type='GENERATION')")

    except ImportError:
        logger.warning("⚠️ SwisperStudio SDK not installed - observability disabled")
        logger.warning("   Install with: pip install swisper-studio-sdk==0.5.2")
    except Exception as e:
        logger.warning(f"⚠️ SwisperStudio observability initialization failed: {e}")
        logger.warning("   Continuing without observability")

# Close SwisperStudio Redis publisher
if settings.SWISPER_STUDIO_ENABLED:
    try:
        from swisper_studio_sdk import close_redis_publisher
        await close_redis_publisher()
        logger.info("✅ SwisperStudio Redis publisher closed")
    except Exception as e:
        logger.warning(f"⚠️ Failed to close SwisperStudio publisher: {e}")

# Start your backend
fastapi run app/main.py

# Check logs - should see:
# ✅ SwisperStudio observability initialized (Redis Streams)
# ✅ LLM prompt capture enabled (structured + streaming)

_tools_executed: List[Dict[str, Any]]  # STANDARD FORMAT: Tool execution trace
_tools_executed_by: Optional[str]      # OWNERSHIP MARKER: Which node created this

class ResearchAgentState(TypedDict):
    # ... existing fields ...
    tool_execution_results_history: List[ToolExecutionResult]  # Backwards compat

    # NEW: SwisperStudio observability
    _tools_executed: List[Dict[str, Any]]
    _tools_executed_by: Optional[str]

    iteration_count: int

# Run type checker
cd backend
mypy app/api/services/agents/

# Should pass without errors (or same errors as before)

# ❌ OLD: Direct StateGraph
workflow = StateGraph(AgentState)

# ✅ NEW: Traced graph with fallback
try:
    from swisper_studio_sdk import create_traced_graph
    workflow = create_traced_graph(
        AgentState,
        trace_name="agent_name"  # Use descriptive name
    )
    self.logger.info("✅ Agent graph wrapped with SwisperStudio tracing")
except ImportError:
    workflow = StateGraph(AgentState)
    self.logger.warning("⚠️ SwisperStudio SDK not available - tracing disabled")

# Start backend
fastapi run app/main.py

# Send a test message that triggers agents
# Check logs for:
# ✅ GlobalSupervisor graph wrapped with SwisperStudio tracing
# ✅ ResearchAgent graph wrapped with SwisperStudio tracing
# ✅ ProductivityAgent graph wrapped with SwisperStudio tracing
# ✅ WealthAgent graph wrapped with SwisperStudio tracing
# ✅ DocumentSearchAgent graph wrapped with SwisperStudio tracing

# In agent build_graph() method:
workflow.add_node("tool_execution", self.tool_execution_node.execute)  # Node name

# In tool_execution_node.py:
state["_tools_executed_by"] = "tool_execution"  # ✅ MUST MATCH!

# In agent build_graph():
workflow.add_node("tool_execution", self.tool_execution_node.execute)

# In tool_execution_node.py:
state["_tools_executed_by"] = "wealth_tool_execution"  # ❌ MISMATCH! Tools won't appear!

# STEP 1: Execute tool (your existing logic)
result = await execute_tool(params)

# STEP 2: Create standardized tool entry
tool_entry = {
    "tool_name": "search_web",           # Tool/function name
    "parameters": {"query": "..."},      # Input parameters
    "result": result,                    # Execution result
    "timestamp": datetime.now(timezone.utc).isoformat(),  # ISO 8601
    "status": "success" if success else "failure"
}

# STEP 3: Append to _tools_executed
existing_tools_executed = state.get("_tools_executed", [])
updated_tools_executed = existing_tools_executed + [tool_entry]

# STEP 4: Update state with ownership marker
# ⚠️ CRITICAL: Use the EXACT node name from add_node()
state["_tools_executed"] = updated_tools_executed
state["_tools_executed_by"] = "tool_execution"  # MUST match add_node("tool_execution", ...)

# Verify all ownership markers match node names
for agent in research_agent productivity_agent wealth_agent doc_agent; do
  echo "=== $agent ==="
  grep "add_node.*tool" apps/backend/swisper/agents/*$agent*/*.py 2>/dev/null
  grep "_tools_executed_by" apps/backend/swisper/agents/*$agent*/nodes/*tool*.py 2>/dev/null
  echo
done
# Node names and ownership values should match for each agent

# Send a message that triggers tools
# Example: "Search for Python tutorials"

# Check Redis for tool execution events
redis-cli
XLEN observability:events
# Should show events > 0

# Check SwisperStudio UI for tool traces (if running)

from unittest.mock import patch
from langgraph.graph import StateGraph

# Mock SwisperStudio SDK to prevent import errors in tests
with patch("swisper_studio_sdk.create_traced_graph") as mock_traced_graph:
    # Make mock return a standard StateGraph
    mock_traced_graph.side_effect = lambda state_class, trace_name: StateGraph(state_class)

    # Your test logic here
    supervisor = GlobalSupervisor(...)
    result = await supervisor.run(...)

# Update Docker container with latest code
docker compose cp apps/backend/swisper/. backend:/app/swisper/
docker compose cp apps/backend/tests/. backend:/app/tests/

# Run tests
docker compose exec backend pytest -vv

# All tests should pass

ImportError: No module named 'swisper_studio_sdk'

# 1. Check if SDK is installed
pip list | grep swisper-studio-sdk

# 2. If not found, install SDK from PyPI
pip install swisper-studio-sdk==0.5.2

# 3. Verify import works
python -c "from swisper_studio_sdk import create_traced_graph; print('✅ SDK works!')"

⚠️ SwisperStudio observability initialization failed: Connection refused

# Check Redis is running
redis-cli ping
# Should return: PONG

# If not running
docker compose up -d redis

# Check Redis URL in config
python -c "from app.core.config import settings; print(settings.SWISPER_STUDIO_REDIS_URL)"
# Should match your Redis host

# Check events are in Redis
redis-cli
XLEN observability:events
# Should be > 0

# Check SwisperStudio consumer is running
docker ps | grep swisper_studio

# Check consumer logs
docker logs swisper_studio_consumer

# Verify project ID matches
# Config project ID should match SwisperStudio project

ImportError: cannot import name 'create_traced_graph' from 'swisper_studio_sdk'

# Add mock to test file
from unittest.mock import patch
from langgraph.graph import StateGraph

with patch("swisper_studio_sdk.create_traced_graph") as mock:
    mock.side_effect = lambda state, name: StateGraph(state)
    # Your test code

KeyError: '_tools_executed'

# Always use .get() with default
existing = state.get("_tools_executed", [])  # ✅ Safe
# NOT: state["_tools_executed"]  # ❌ Crashes if missing

# 1. Find the node name in agent build_graph() method
grep "add_node.*tool" apps/backend/swisper/agents/your_agent/your_agent.py

# Example output:
# workflow.add_node("tool_execution", self.tool_execution_node.execute)
#                    ^^^^^^^^^^^^^^^
#                    This is your node name!

# 2. Update ownership marker in tool execution node to MATCH
# In apps/backend/swisper/agents/your_agent/nodes/your_tool_execution_node.py:
state["_tools_executed_by"] = "tool_execution"  # ✅ Must match node name above!

# Check if node names and ownership match
grep "add_node.*tool" apps/backend/swisper/agents/*/your_agent.py
grep "_tools_executed_by" apps/backend/swisper/agents/*/nodes/*tool*.py

# Node name and ownership value should be IDENTICAL

# ❌ WRONG:
workflow.add_node("tool_execution", ...)
state["_tools_executed_by"] = "wealth_tool_execution"  # Mismatch!

# ✅ CORRECT:
workflow.add_node("tool_execution", ...)
state["_tools_executed_by"] = "tool_execution"  # Match!

# ✅ ALSO CORRECT (if node has unique name):
workflow.add_node("doc_tool_execution_node", ...)
state["_tools_executed_by"] = "doc_tool_execution_node"  # Match!

# Checked Redis stream - all observation_end events show:
observation_end type distribution (500 events):
  SPAN: 40
  NONE: 5  
  AGENT: 1
  GENERATION: 0  ❌ <-- No LLM types being sent!

# Force GENERATION type for nodes that always make LLM calls
from swisper_studio_sdk import traced

@traced(name="classify_intent", observation_type="GENERATION")  # Force LLM type
async def classify_intent(state):
    """This function makes an LLM call, force GENERATION type."""
    result = await llm.get_structured_output(...)
    return result

# Add debug logging:
@traced(name="classify_intent", observation_type="GENERATION")
async def classify_intent(state):
    logger.info("🔍 classify_intent: Starting LLM call")
    try:
        result = await llm_adapter.get_structured_output(...)
        logger.info(f"✅ classify_intent: LLM result received")
        return result
    except Exception as e:
        logger.error(f"❌ classify_intent: LLM call failed: {e}")
        raise

# observability_consumer.py lines 363-366:
if "type" in data and data.get("type"):
    observation.type = data.get("type")

from swisper_studio_sdk import traced

from swisper_studio_sdk import traced

@traced(name="my_function")
async def my_function(state):
    # Your logic here
    result = await some_operation()
    return result

@traced(name="classify_intent", observation_type="GENERATION")
async def classify_intent(state):
    """Force GENERATION type for this LLM-calling function."""
    # LLM call happens here
    result = await llm.get_structured_output(...)
    return result

@traced(name="process_request")
async def process_request(state):
    """Parent observation."""

    # Child observation (nested under process_request)
    result = await classify_intent(state)

    return result

@traced(name="classify_intent", observation_type="GENERATION")
async def classify_intent(state):
    """Child observation - will appear nested in UI."""
    return await llm.classify(state)

@traced(name="transform_data", capture_input=True, capture_output=True)
async def transform_data(input_state):
    """
    Both input and output state will be captured.
    Useful for debugging state transformations.
    """
    output_state = {**input_state, "transformed": True}
    return output_state

Phase 1: observation_start
├── Type: "SPAN" (placeholder)
├── Sent to SwisperStudio
└── Observation created in DB

Phase 2: observation_end  
├── Type: "GENERATION" (if LLM data captured) OR "SPAN" (default)
├── Sent to SwisperStudio
└── Observation type UPDATED in DB

# Inside SDK @traced decorator:
def _detect_observation_type(name, has_llm_data, has_tool_data):
    if has_llm_data:
        return "GENERATION"  # LLM call detected!
    elif has_tool_data:
        return "TOOL"
    elif name in KNOWN_AGENT_NAMES:
        return "AGENT"
    else:
        return "SPAN"  # Default fallback

# Instead of:
@traced(name="classify_intent")  # ❌ May detect as SPAN

# Use:
@traced(name="classify_intent", observation_type="GENERATION")  # ✅ Force LLM type

# In Swisper backend, add to your LLM-calling function:
import logging
logger = logging.getLogger(__name__)

@traced(name="classify_intent", observation_type="GENERATION")
async def classify_intent(state):
    logger.info("📤 classify_intent: observation_type=GENERATION (forced)")
    result = await llm.get_structured_output(...)
    logger.info(f"📥 classify_intent: result received")
    return result

# apps/backend/swisper/services/observability_consumer.py, lines 363-366:
# Update type if provided (for AUTO type detection)
if "type" in data and data.get("type"):
    observation.type = data.get("type")

_tools_executed: List[Dict[str, Any]]  # STANDARD FORMAT: Tool execution trace
_tools_executed_by: Optional[str]      # OWNERSHIP MARKER: Which node created this

{
    "tool_name": str,           # Tool/function name
    "parameters": dict,         # Input parameters
    "result": Any,              # Execution result
    "timestamp": str,           # ISO 8601 timestamp
    "status": str               # "success" | "failure"
}

class GlobalSupervisorState(TypedDict):
    # ... existing fields ...

    # NEW: SwisperStudio observability fields
    _tools_executed: Optional[List[Dict[str, Any]]]
    _tools_executed_by: Optional[str]

class ResearchAgentState(TypedDict):
    # ... existing fields ...
    tool_execution_results_history: List[ToolExecutionResult]  # Backwards compat

    # NEW: SwisperStudio observability
    _tools_executed: List[Dict[str, Any]]
    _tools_executed_by: Optional[str]

    iteration_count: int

class ProductivityAgentState(TypedDict):
    # ... existing fields ...
    tool_results: Dict[str, Any]  # Backwards compat format

    # NEW: SwisperStudio observability
    _tools_executed: List[Dict[str, Any]]
    _tools_executed_by: Optional[str]

    final_result: Optional[str]

class WealthAgentState(TypedDict):
    # ... existing fields ...
    tool_execution_result: Optional[Any]  # Backwards compat

    # NEW: SwisperStudio observability
    _tools_executed: List[Dict[str, Any]]
    _tools_executed_by: Optional[str]

    wealth_planner_result: Optional[Any]

class DocumentSearchAgentState(TypedDict):
    # ... existing fields ...
    tool_results: Dict[str, str]  # Backwards compat format

    # NEW: SwisperStudio observability
    _tools_executed: List[Dict[str, Any]]
    _tools_executed_by: Optional[str]

    tools: List[DocAgentToolInfo]
    available_documents: List[DocumentInfo]

# ❌ OLD: Direct StateGraph
workflow = StateGraph(AgentState)

# ✅ NEW: Traced graph with fallback
try:
    from swisper_studio_sdk import create_traced_graph
    workflow = create_traced_graph(
        AgentState,
        trace_name="agent_name"
    )
    logger.info("✅ Agent graph wrapped with SwisperStudio tracing")
except ImportError:
    workflow = StateGraph(AgentState)
    logger.warning("⚠️ SwisperStudio SDK not available - tracing disabled")

# Initialize graph with SwisperStudio tracing (if SDK available)
try:
    from swisper_studio_sdk import create_traced_graph
    graph = create_traced_graph(
        GlobalSupervisorState,
        trace_name="global_supervisor"
    )
    self.logger.info("✅ GlobalSupervisor graph wrapped with SwisperStudio tracing")
except ImportError:
    # Fallback to standard StateGraph if SDK not available
    graph = StateGraph(GlobalSupervisorState)
    self.logger.debug("Using standard StateGraph (SwisperStudio SDK not available)")

# Create workflow graph with SwisperStudio tracing
try:
    from swisper_studio_sdk import create_traced_graph
    workflow = create_traced_graph(
        ResearchAgentState,
        trace_name="research_agent"
    )
    self.logger.info("✅ ResearchAgent graph wrapped with SwisperStudio tracing")
except ImportError:
    workflow = StateGraph(ResearchAgentState)
    self.logger.warning("⚠️ SwisperStudio SDK not available - ResearchAgent tracing disabled")

# Create workflow graph with optimized state and SwisperStudio tracing
try:
    from swisper_studio_sdk import create_traced_graph
    workflow = create_traced_graph(
        ProductivityAgentState,
        trace_name="productivity_agent"
    )
    self.logger.info("✅ ProductivityAgent graph wrapped with SwisperStudio tracing")
except ImportError:
    workflow = StateGraph(ProductivityAgentState)
    self.logger.warning("⚠️ SwisperStudio SDK not available - ProductivityAgent tracing disabled")

# Create workflow graph with SwisperStudio tracing
try:
    from swisper_studio_sdk import create_traced_graph
    workflow = create_traced_graph(
        WealthAgentState,
        trace_name="wealth_agent"
    )
    self.logger.info("✅ WealthAgent graph wrapped with SwisperStudio tracing")
except ImportError:
    workflow = StateGraph(WealthAgentState)
    self.logger.warning("⚠️ SwisperStudio SDK not available - WealthAgent tracing disabled")

# Create workflow graph with SwisperStudio tracing
try:
    from swisper_studio_sdk import create_traced_graph
    workflow = create_traced_graph(
        DocumentSearchAgentState,
        trace_name="document_search_agent"
    )
    self.logger.info("✅ DocumentSearchAgent graph wrapped with SwisperStudio tracing")
except ImportError:
    workflow = StateGraph(DocumentSearchAgentState)
    self.logger.warning("⚠️ SwisperStudio SDK not available - DocumentSearchAgent tracing disabled")

# STEP 1: Execute tool (existing logic)
result = await execute_tool(params)

# STEP 2: Create standardized tool entry
tool_entry = {
    "tool_name": "tool_name",
    "parameters": params,
    "result": result,
    "timestamp": datetime.now(timezone.utc).isoformat(),
    "status": "success" if success else "failure"
}

# STEP 3: Append to _tools_executed
existing_tools_executed = state.get("_tools_executed", [])
updated_tools_executed = existing_tools_executed + [tool_entry]

# STEP 4: Update state with ownership marker
state["_tools_executed"] = updated_tools_executed
state["_tools_executed_by"] = "node_name"  # Prevents duplicate extraction

# STANDARDIZATION: Initialize _tools_executed for new standard format
tools_executed_standard = []

# ... execute tools ...

# Add to standard format
tools_executed_standard.append({
    "tool_name": operation.tool_name,
    "parameters": operation.parameters,
    "result": tool_result,
    "timestamp": datetime.now(timezone.utc).isoformat(),
    "status": "success" if success else "failure"
})

# Get existing _tools_executed or initialize
existing_tools_executed = state.get("_tools_executed", [])
updated_tools_executed = existing_tools_executed + tools_executed_standard

return {
    **state,
    "tool_execution_results_history": updated_history,
    "_tools_executed": updated_tools_executed,  # NEW STANDARD FORMAT
    "_tools_executed_by": "tool_execution"  # OWNERSHIP MARKER
}

"""
STANDARDIZATION (v0.5.2): Now populates both formats:
- tool_results: Dict (backwards compat)
- _tools_executed: List[Dict] (new standard)
"""

# STANDARDIZATION: Convert to new _tools_executed format
existing_tools_executed = state.get("_tools_executed", [])
new_tools_executed = []

for result_key, result_data in execution_results.items():
    if isinstance(result_data, dict):
        new_tools_executed.append({
            "tool_name": result_data.get("tool_name", "unknown"),
            "parameters": result_data.get("tool_args", {}),
            "result": result_data.get("result"),
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "status": "success" if result_data.get("success") else "failure"
        })

updated_tools_executed = existing_tools_executed + new_tools_executed

updated_state = {
    **state,
    "tool_results": updated_tool_results,
    "_tools_executed": updated_tools_executed,  # NEW STANDARD FORMAT
    "_tools_executed_by": "tool_execution",  # OWNERSHIP MARKER
}

# STANDARDIZATION (v0.5.2): Populate _tools_executed format
tools_executed_standard = []
for operation in wealth_planner_result.tool_operations:
    tool_result = results.get(operation.operation_id)
    success = tool_result and not isinstance(tool_result, Exception)

    tools_executed_standard.append({
        "tool_name": operation.tool_name,
        "parameters": operation.parameters,
        "result": str(tool_result) if tool_result else None,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "status": "success" if success else "failure"
    })

existing_tools_executed = state.get("_tools_executed", [])
updated_tools_executed = existing_tools_executed + tools_executed_standard

return {
    **state,
    "tool_execution_result": tool_execution_result,
    "_tools_executed": updated_tools_executed,  # NEW STANDARD FORMAT
    "_tools_executed_by": "wealth_tool_execution",  # OWNERSHIP MARKER
}

"""
STANDARDIZATION (v0.5.2): Populates both tool_results (backwards compat)
and _tools_executed (new standard format).
"""

# Execute tool (existing logic)
result = await tool_executor(current_plan.tool_input)

# Create standardized tool entry
tool_entry = {
    "tool_name": current_plan.tool,
    "parameters": {"input": current_plan.tool_input},
    "result": result,
    "timestamp": datetime.now(timezone.utc).isoformat(),
    "status": "success" if result else "failure"
}

# STANDARDIZATION: Add to _tools_executed
existing_tools_executed = state.get("_tools_executed", [])
updated_tools_executed = existing_tools_executed + [tool_entry]
state["_tools_executed"] = updated_tools_executed
state["_tools_executed_by"] = "doc_tool_execution"  # OWNERSHIP MARKER

# SwisperStudio Integration (Observability & Configuration Management)
# SDK v0.5.2: Redis Streams-based observability (50x faster than HTTP)
SWISPER_STUDIO_ENABLED: bool = True
SWISPER_STUDIO_REDIS_URL: str = "redis://redis:6379"  # Redis for observability events
SWISPER_STUDIO_STREAM_NAME: str = "observability:events"
SWISPER_STUDIO_PROJECT_ID: str = "0d7aa606-cb29-4a31-8a59-50fa61151a32"
SWISPER_STUDIO_CAPTURE_REASONING: bool = True  # Capture LLM reasoning (<think> tags)
SWISPER_STUDIO_REASONING_MAX_LENGTH: int = 50000  # 50 KB limit for reasoning

# Initialize SwisperStudio observability (SDK v0.5.2 - Q2: Tracing Toggle)
if settings.SWISPER_STUDIO_ENABLED:
    try:
        from swisper_studio_sdk import initialize_redis_publisher, wrap_llm_adapter

        # Initialize Redis Streams publisher for observability
        await initialize_redis_publisher(
            redis_url=settings.SWISPER_STUDIO_REDIS_URL,
            stream_name=settings.SWISPER_STUDIO_STREAM_NAME,
            project_id=settings.SWISPER_STUDIO_PROJECT_ID,
            verify_consumer=True,  # Check if SwisperStudio consumer is running
        )
        logger.info("✅ SwisperStudio observability initialized (Redis Streams)")
        logger.info(f"   Redis: {settings.SWISPER_STUDIO_REDIS_URL}")
        logger.info(f"   Stream: {settings.SWISPER_STUDIO_STREAM_NAME}")
        logger.info(f"   Project ID: {settings.SWISPER_STUDIO_PROJECT_ID}")

        # Enable LLM prompt capture (includes streaming support)
        try:
            wrap_llm_adapter()
            logger.info("✅ LLM prompt capture enabled (structured + streaming)")
        except Exception as e:
            logger.warning(f"⚠️ LLM prompt capture disabled: {e}")
            logger.warning("   State capture will still work")

    except ImportError:
        logger.warning("⚠️ SwisperStudio SDK not installed - observability disabled")
        logger.warning("   Install with: pip install swisper-studio-sdk==0.5.2")
    except Exception as e:
        logger.warning(f"⚠️ SwisperStudio observability initialization failed: {e}")
        logger.warning("   Continuing without observability - events will queue in Redis")

# Close SwisperStudio Redis publisher
if settings.SWISPER_STUDIO_ENABLED:
    try:
        from swisper_studio_sdk import close_redis_publisher
        await close_redis_publisher()
        logger.info("✅ SwisperStudio Redis publisher closed")
    except Exception as e:
        logger.warning(f"⚠️ Failed to close SwisperStudio publisher: {e}")

# Mock SwisperStudio SDK to prevent import errors in tests
with patch("swisper_studio_sdk.create_traced_graph") as mock_traced_graph:
    # Make mock return a standard StateGraph
    mock_traced_graph.side_effect = lambda state_class, trace_name: StateGraph(state_class)

    # Your test logic here
    supervisor = GlobalSupervisor(...)
    result = await supervisor.run(...)

return {
    **state,
    # LEGACY: Backward compatible format
    "tool_results": updated_tool_results,

    # NEW: Standard observability format
    "_tools_executed": updated_tools_executed,
    "_tools_executed_by": "tool_execution"
}

state["_tools_executed_by"] = "tool_execution"  # Mark ownership

# In SDK decorator (internal):
created_by = output.get('_tools_executed_by')
is_owner = (created_by == node_name)  # EXACT match required!

if is_owner:
    # Extract tools and create wrench icons 🔧
else:
    # Skip tool extraction (no wrench icons!)

grep "add_node.*tool" apps/backend/swisper/agents/wealth/agent.py
# Output: workflow.add_node("tool_execution", self.tool_execution_node.execute)
#                              ^^^^^^^^^^^^^^^
#                              Your node name

grep "_tools_executed_by" apps/backend/swisper/agents/wealth/nodes/wealth_tool_execution_node.py
# Should show: state["_tools_executed_by"] = "tool_execution"  # MUST MATCH!

# Install from PyPI (no authentication needed)
pip install swisper-studio-sdk==0.5.2

# Method 1: From Git + Tag (if you need unreleased changes)
pip install git+https://github.com/Fintama/swisper_studio.git@sdk-v0.5.2#subdirectory=sdk

# Method 2: From local clone (for SDK development)
pip install -e /path/to/swisper_studio/sdk

# Method 3: In requirements.txt
# swisper-studio-sdk==0.5.2

cd backend
pip install -e /path/to/swisper_studio/sdk

# Or with uv
uv pip install -e /path/to/swisper_studio/sdk

┌─────────────────────────────────────────────────────────────┐
│                    FastAPI Application                      │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐  │
│  │              Startup (main.py)                       │  │
│  │  1. Initialize Redis publisher                       │  │
│  │  2. Wrap LLM adapter                                 │  │
│  │  3. Verify consumer running                          │  │
│  └─────────────────────────────────────────────────────┘  │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐  │
│  │           Agents (LangGraph Workflows)               │  │
│  │                                                       │  │
│  │  ┌─────────────────────────────────────────────┐    │  │
│  │  │  Global Supervisor                           │    │  │
│  │  │  • create_traced_graph(GlobalSupervisorState)│   │  │
│  │  │  • _tools_executed: aggregated from domains  │    │  │
│  │  └─────────────────────────────────────────────┘    │  │
│  │                                                       │  │
│  │  ┌─────────────────────────────────────────────┐    │  │
│  │  │  Domain Agents                               │    │  │
│  │  │  • create_traced_graph(AgentState)           │    │  │
│  │  │  • Tool nodes populate _tools_executed       │    │  │
│  │  │  • Ownership marker: _tools_executed_by      │    │  │
│  │  └─────────────────────────────────────────────┘    │  │
│  └─────────────────────────────────────────────────────┘  │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐  │
│  │         SwisperStudio SDK (v0.5.2)                   │  │
│  │  • create_traced_graph() - Wraps StateGraph         │  │
│  │  • Captures state transitions                        │  │
│  │  • Extracts _tools_executed                          │  │
│  │  • Publishes to Redis Streams                        │  │
│  └─────────────────────────────────────────────────────┘  │
│                           │                                │
└───────────────────────────┼────────────────────────────────┘
                            │
                            ▼
                  ┌──────────────────┐
                  │  Redis Streams   │
                  │  observability:  │
                  │     events       │
                  └──────────────────┘
                            │
                            ▼
                  ┌──────────────────┐
                  │ SwisperStudio UI │
                  │  (Consumer)      │
                  └──────────────────┘

pip install swisper-studio-sdk==0.5.2

from swisper_studio_sdk import (
    create_traced_graph,      # Wrap LangGraph StateGraph
    traced,                   # Decorator for individual functions
    initialize_redis_publisher,  # Start Redis publisher
    close_redis_publisher,    # Stop Redis publisher
    wrap_llm_adapter,         # Enable LLM telemetry capture
)

# Basic tracing
@traced(name="my_function")

# Force LLM type (for LLM-calling functions)
@traced(name="classify_intent", observation_type="GENERATION")

# Tool execution
@traced(name="execute_search", observation_type="TOOL")

# Agent orchestration
@traced(name="supervisor", observation_type="AGENT")

# Disable input/output capture (for sensitive data)
@traced(name="process_secrets", capture_input=False, capture_output=False)

from swisper_studio_sdk import create_traced_graph

workflow = create_traced_graph(MyAgentState, trace_name="my_agent")