cameron.stream
a digital person for bluesky
1#!/usr/bin/env python
2"""Quick CLI tool to send a message to void and stream the response."""
3
4import os
5import sys
6from dotenv import load_dotenv
7from letta_client import Letta
8import argparse
9
10def send_message_to_void(message: str):
11 """Send a message to void and stream the response."""
12 load_dotenv()
13
14 # Create Letta client (v1.0: token → api_key)
15 client = Letta(
16 base_url=os.getenv("LETTA_BASE_URL", "http://localhost:8283"),
17 api_key=os.getenv("LETTA_API_KEY")
18 )
19
20 # Get the void agent (v1.0: list returns page object)
21 agents_page = client.agents.list()
22 agents = agents_page.items if hasattr(agents_page, 'items') else agents_page
23 void_agent = next((a for a in agents if a.name == "void"), None)
24
25 if not void_agent:
26 print("Error: void agent not found")
27 return
28
29 print(f"Sending message to void: {message}\n")
30 print("=" * 50)
31
32 # Send message and stream response
33 try:
34 # Use the streaming interface (v1.0: create_stream → stream)
35 message_stream = client.agents.messages.stream(
36 agent_id=void_agent.id,
37 messages=[{"role": "user", "content": message}],
38 stream_tokens=False, # Step streaming only
39 max_steps=100
40 )
41
42 # Process the streaming response
43 for chunk in message_stream:
44 if hasattr(chunk, 'message_type'):
45 if chunk.message_type == 'reasoning_message':
46 # Show reasoning
47 print("\n◆ Reasoning")
48 print(" ─────────")
49 for line in chunk.reasoning.split('\n'):
50 print(f" {line}")
51 elif chunk.message_type == 'tool_call_message':
52 # Show tool calls
53 tool_name = chunk.tool_call.name if hasattr(chunk, 'tool_call') else 'unknown'
54 print(f"\n▸ Calling tool: {tool_name}")
55 elif chunk.message_type == 'tool_return_message':
56 # Show tool results
57 if hasattr(chunk, 'tool_return') and chunk.tool_return:
58 result_str = str(chunk.tool_return)
59 print(f" Tool result: {result_str[:200]}...")
60 elif chunk.message_type == 'assistant_message':
61 # Show assistant response
62 print("\n▶ Assistant Response")
63 print(" ──────────────────")
64 for line in chunk.content.split('\n'):
65 print(f" {line}")
66 elif chunk.message_type not in ['usage_statistics', 'stop_reason']:
67 # Filter out verbose message types
68 print(f" {chunk.message_type}: {str(chunk)[:150]}...")
69
70 if str(chunk) == 'done':
71 break
72
73 print("\n" + "=" * 50)
74
75 except Exception as e:
76 print(f"Error: {e}")
77 import traceback
78 traceback.print_exc()
79
80def main():
81 parser = argparse.ArgumentParser(description="Send a quick message to void")
82 parser.add_argument("message", nargs="+", help="Message to send to void")
83 args = parser.parse_args()
84
85 # Join the message parts
86 message = " ".join(args.message)
87
88 send_message_to_void(message)
89
90if __name__ == "__main__":
91 main()