xray.trace()

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

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

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

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

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