1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
|
#!/usr/bin/env python3
"""
Debug version of streaming chat that shows API errors in detail.
"""
import asyncio
import sys
import os
import json
from datetime import datetime
from typing import Dict, Any
sys.path.insert(0, 'src')
from claude import ClaudeAgentClient, AgentOptions, ToolResult
from claude.streaming import StreamParser
import httpx
# ANSI color codes
class Colors:
USER = '\033[1;36m' # Cyan
ASSISTANT = '\033[1;32m' # Green
TOOL = '\033[1;33m' # Yellow
ERROR = '\033[1;31m' # Red
STAT = '\033[0;90m' # Gray
RESET = '\033[0m'
async def test_simple_message():
"""Test a simple message first."""
print("=" * 70)
print(f"{Colors.ASSISTANT}Testing Simple Message (No Tools){Colors.RESET}")
print("=" * 70)
options = AgentOptions(
model="claude-sonnet-4-5-20250929",
stream=True,
allowed_tools=[] # No tools for this test
)
client = ClaudeAgentClient(options=options)
try:
print(f"\n{Colors.USER}You:{Colors.RESET} Hello Claude")
print(f"\n{Colors.ASSISTANT}Assistant:{Colors.RESET} ", end='', flush=True)
async for chunk in client.send_message_stream("Hello Claude, please respond with a short greeting."):
if chunk.text_delta:
print(chunk.text_delta, end='', flush=True)
print()
print(f"\n{Colors.ASSISTANT}✅ Simple message works!{Colors.RESET}\n")
except httpx.HTTPStatusError as e:
print(f"\n\n{Colors.ERROR}❌ HTTP Error: {e.response.status_code}{Colors.RESET}")
print(f"{Colors.ERROR}Response: {e.response.text}{Colors.RESET}\n")
return False
except Exception as e:
print(f"\n\n{Colors.ERROR}❌ Error: {e}{Colors.RESET}\n")
import traceback
traceback.print_exc()
return False
finally:
await client.close()
return True
async def test_with_tools():
"""Test with tools enabled."""
print("=" * 70)
print(f"{Colors.ASSISTANT}Testing With Tools Enabled{Colors.RESET}")
print("=" * 70)
options = AgentOptions(
model="claude-sonnet-4-5-20250929",
stream=True,
allowed_tools=["Bash"] # Only Bash for this test
)
client = ClaudeAgentClient(options=options)
# Show what we're sending
print(f"\n{Colors.STAT}Request configuration:{Colors.RESET}")
print(f" Model: {options.model}")
print(f" Stream: {options.stream}")
print(f" Tools: {options.allowed_tools}")
filtered_tools = client._filter_tools()
print(f" Filtered tools: {[t['name'] for t in filtered_tools]}")
system_prompt = client._build_system_prompt()
print(f" System prompt blocks: {len(system_prompt)}")
if not system_prompt:
print(f" {Colors.ERROR}⚠️ WARNING: System prompt is empty!{Colors.RESET}")
try:
print(f"\n{Colors.USER}You:{Colors.RESET} Check my Python version")
print(f"\n{Colors.ASSISTANT}Assistant:{Colors.RESET} ", end='', flush=True)
parser = StreamParser()
async for chunk in client.send_message_stream("Check my Python version"):
parser.add_chunk(chunk)
if chunk.text_delta:
print(chunk.text_delta, end='', flush=True)
print()
# Check for tool uses
complete_message = parser.to_dict()
content_blocks = complete_message.get("content", [])
tool_uses = [block for block in content_blocks if block.get("type") == "tool_use"]
if tool_uses:
print(f"\n{Colors.TOOL}Claude wants to use {len(tool_uses)} tool(s){Colors.RESET}")
for tool_use in tool_uses:
print(f"{Colors.TOOL}[Tool: {tool_use.get('name')}]{Colors.RESET}")
print(f"{Colors.STAT} Input: {json.dumps(tool_use.get('input', {}), indent=2)}{Colors.RESET}")
else:
print(f"\n{Colors.STAT}No tools used in response{Colors.RESET}")
print(f"\n{Colors.ASSISTANT}✅ Request with tools succeeded!{Colors.RESET}\n")
except httpx.HTTPStatusError as e:
print(f"\n\n{Colors.ERROR}❌ HTTP Error: {e.response.status_code}{Colors.RESET}")
print(f"{Colors.ERROR}Response body:{Colors.RESET}")
try:
error_json = e.response.json()
print(json.dumps(error_json, indent=2))
except:
print(e.response.text)
# Show what we tried to send
print(f"\n{Colors.ERROR}Request that failed:{Colors.RESET}")
print(f" Messages: {len(client.messages)}")
print(f" System: {len(system_prompt)} blocks")
print(f" Tools: {len(filtered_tools)}")
return False
except Exception as e:
print(f"\n\n{Colors.ERROR}❌ Error: {e}{Colors.RESET}\n")
import traceback
traceback.print_exc()
return False
finally:
await client.close()
return True
async def main():
"""Run debug tests."""
# Check for API key
if not os.getenv('ANTHROPIC_API_KEY'):
print(f"{Colors.ERROR}❌ Error: ANTHROPIC_API_KEY environment variable not set{Colors.RESET}")
print("\nSet your API key:")
print(" export ANTHROPIC_API_KEY='sk-ant-...'")
return 1
print(f"\n{Colors.ASSISTANT}Claude API Debug Tool{Colors.RESET}\n")
# Test 1: Simple message without tools
success1 = await test_simple_message()
if not success1:
print(f"{Colors.ERROR}Simple message failed - stopping here{Colors.RESET}")
return 1
# Test 2: With tools enabled
success2 = await test_with_tools()
if success2:
print(f"\n{Colors.ASSISTANT}✅ All tests passed!{Colors.RESET}\n")
return 0
else:
print(f"\n{Colors.ERROR}❌ Tool test failed{Colors.RESET}\n")
return 1
if __name__ == "__main__":
try:
exit_code = asyncio.run(main())
sys.exit(exit_code)
except KeyboardInterrupt:
print(f"\n\n{Colors.ASSISTANT}⚠️ Interrupted by user{Colors.RESET}")
sys.exit(130)
except Exception as e:
print(f"\n{Colors.ERROR}❌ Fatal error: {e}{Colors.RESET}")
import traceback
traceback.print_exc()
sys.exit(1)
|
