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