xray.trace()

1from connectonion import Agent
2from connectonion.decorators import xray
3
4@xray
5def greet(name: str) -> str:
6    """Simple greeting function."""
7    return f"Hello, {name}!"
8
9# Create agent
10agent = Agent("greeter", tools=[greet])
11result = agent.input("Say hello to Alice")
12
13# View execution trace
14xray.trace(agent)

1from connectonion import Agent
2from connectonion.decorators import xray
3
4@xray
5def analyze_text(text: str) -> dict:
6    """Analyze text and return metrics."""
7    return {
8        'word_count': len(text.split()),
9        'char_count': len(text),
10        'sentiment': 'neutral'
11    }
12
13@xray
14def generate_summary(text: str, max_words: int = 10) -> str:
15    """Generate a summary of the text."""
16    words = text.split()[:max_words]
17    return f"Summary: {' '.join(words)}..."
18
19# Create and run agent
20agent = Agent(
21    name="text_processor",
22    tools=[analyze_text, generate_summary]
23)
24
25result = agent.input("Analyze this sample text and create a summary")
26
27# View the execution trace
28xray.trace(agent)

1from connectonion import Agent
2from connectonion.decorators import xray
3
4@xray
5def validate_email(email: str) -> dict:
6    """Validate email format."""
7    if '@' not in email:
8        raise ValueError("Invalid email: missing @ symbol")
9    parts = email.split('@')
10    if len(parts) != 2 or not parts[0] or not parts[1]:
11        raise ValueError("Invalid email format")
12    return {"email": email, "valid": True}
13
14@xray
15def send_notification(email_data: dict) -> str:
16    """Send notification to validated email."""
17    if not email_data.get("valid"):
18        return "❌ Cannot send to invalid email"
19    return f"✅ Notification sent to {email_data['email']}"
20
21@xray
22def fallback_contact(message: str) -> str:
23    """Fallback contact method when email fails."""
24    return f"📞 Fallback: {message} (via phone/SMS)"
25
26# Create agent
27agent = Agent(
28    name="notification_agent",
29    tools=[validate_email, send_notification, fallback_contact]
30)
31
32result = agent.input("Send notification to user at bad-email-format")
33
34# View trace showing error handling
35xray.trace(agent)

1from connectonion import Agent
2from connectonion.decorators import xray
3import time
4import random
5
6@xray
7def fetch_user_data(user_id: str) -> dict:
8    """Fast database lookup."""
9    time.sleep(0.05)  # 50ms DB query
10    return {"user_id": user_id, "name": "John Doe", "tier": "premium"}
11
12@xray
13def complex_calculation(data: dict) -> dict:
14    """CPU intensive calculation."""
15    time.sleep(1.2)  # 1.2s processing
16    score = random.randint(750, 850)
17    return {"score": score, "calculation_time": "1.2s", "tier": data["tier"]}
18
19@xray
20def external_api_call(user_data: dict, score_data: dict) -> dict:
21    """Slow external API call."""
22    time.sleep(0.8)  # 800ms network call
23    return {
24        "recommendation": f"Premium package for {user_data['name']}",
25        "api_response_time": "800ms",
26        "confidence": 0.95
27    }
28
29@xray
30def format_response(user_data: dict, score_data: dict, recommendation: dict) -> str:
31    """Fast formatting step."""
32    time.sleep(0.01)  # 10ms formatting
33    return f"""
34    🎯 Analysis Complete:
35    User: {user_data['name']} (ID: {user_data['user_id']})
36    Score: {score_data['score']} ({score_data['tier']} tier)
37    Recommendation: {recommendation['recommendation']}
38    Confidence: {recommendation['confidence']*100:.0f}%
39    """
40
41# Create agent
42agent = Agent(
43    name="analytics_engine",
44    tools=[fetch_user_data, complex_calculation, external_api_call, format_response]
45)
46
47result = agent.input("Generate analytics report for user ID 12345")
48
49# Analyze performance bottlenecks
50xray.trace(agent)

1from connectonion import Agent
2from connectonion.decorators import xray
3
4@xray
5def process_order(order_data: dict) -> dict:
6    """Process customer order with validation."""
7    # 🔴 SET BREAKPOINT HERE - Debug order validation
8    # In IDE debugger console:
9    # >>> xray.trace()  # See current execution state
10    # >>> xray.agent.name  # Current agent name
11    # >>> xray.task  # Original user request
12    # >>> order_data  # Inspect current parameters
13    
14    if not order_data.get("items"):
15        return {"error": "No items in order", "status": "failed"}
16    
17    # Calculate totals
18    total = sum(item.get("price", 0) * item.get("qty", 1) 
19                for item in order_data["items"])
20    
21    # 🔴 BREAKPOINT OPPORTUNITY - Check calculations
22    # >>> xray.trace()  # Updated trace
23    # >>> total  # Inspect calculated total
24    
25    return {
26        "order_id": f"ORD-{hash(str(order_data)) % 10000:04d}",
27        "total": total,
28        "status": "validated",
29        "items_count": len(order_data["items"])
30    }
31
32@xray
33def apply_discounts(validated_order: dict) -> dict:
34    """Apply business rules and discounts."""
35    # 🔴 BREAKPOINT - Discount logic debugging
36    # >>> xray.trace()  # See previous tool results
37    # >>> validated_order  # Inspect order from previous step
38    
39    if validated_order.get("error"):
40        return validated_order  # Pass through errors
41    
42    total = validated_order["total"]
43    discount = 0
44    
45    # Business rules
46    if total > 100:
47        discount = total * 0.1  # 10% discount for orders > $100
48    if validated_order["items_count"] >= 5:
49        discount = max(discount, total * 0.15)  # 15% for bulk orders
50    
51    # 🔴 BREAKPOINT - Final calculations
52    # >>> discount  # Check discount amount
53    # >>> xray.trace()  # See full execution flow
54    
55    final_total = total - discount
56    
57    return {
58        **validated_order,
59        "original_total": total,
60        "discount": discount,
61        "final_total": final_total,
62        "discount_percent": (discount / total * 100) if total > 0 else 0
63    }
64
65@xray
66def generate_invoice(order: dict) -> str:
67    """Generate final invoice."""
68    # 🔴 FINAL BREAKPOINT - Invoice generation
69    # >>> xray.trace()  # Complete execution trace
70    # >>> order  # Final order data
71    
72    if order.get("error"):
73        return f"❌ Order failed: {order['error']}"
74    
75    return f"""
76    🧾 INVOICE #{order['order_id']}
77    ═══════════════════════════
78    Items: {order['items_count']}
79    Subtotal: ${order['original_total']:.2f}
80    Discount ({order['discount_percent']:.1f}%): -${order['discount']:.2f}
81    ───────────────────────────
82    TOTAL: ${order['final_total']:.2f}
83    """
84
85# Create agent for step-by-step debugging
86agent = Agent(
87    name="order_processor",
88    tools=[process_order, apply_discounts, generate_invoice]
89)
90
91# Complex order to debug
92order_data = {
93    "customer": "Alice Smith",
94    "items": [
95        {"name": "Laptop", "price": 899.99, "qty": 1},
96        {"name": "Mouse", "price": 29.99, "qty": 2},
97        {"name": "Keyboard", "price": 79.99, "qty": 1},
98        {"name": "Monitor", "price": 299.99, "qty": 2},
99        {"name": "Cables", "price": 15.99, "qty": 3}
100    ]
101}
102
103# Run with IDE breakpoints set at marked locations
104result = agent.input(f"Process this order: {order_data}")
105
106# Final comprehensive trace
107xray.trace(agent)