#!/usr/bin/env python3
"""
Test script for streaming chat functionality
"""
import asyncio
import requests
import json
async def test_streaming_endpoint():
"""Test the streaming chat endpoint"""
url = "http://localhost:8000/api/chat/stream"
payload = {
"message": "Tell me a short story about a robot learning to paint",
"model": "mistral:latest",
"temperature": 0.7,
"max_tokens": 500
}
print("Testing streaming chat endpoint...")
print(f"Sending: {payload['message']}")
print(f"Model: {payload['model']}")
print("="*60)
try:
response = requests.post(
url,
json=payload,
stream=True,
timeout=30
)
if response.status_code == 200:
print("Connection successful! Streaming response:\n")
full_response = ""
chunk_count = 0
for line in response.iter_lines():
if line:
line_str = line.decode('utf-8')
if line_str.startswith('data: '):
try:
data = json.loads(line_str[6:])
if data.get('content'):
content = data['content']
full_response += content
print(content, end='', flush=True)
chunk_count += 1
if data.get('done'):
print(f"\n\nStreaming completed!")
print(f"Received {chunk_count} chunks")
print(f"Total response length: {len(full_response)} characters")
break
except json.JSONDecodeError:
continue
if not full_response:
print("No content received")
else:
print(f"HTTP Error: {response.status_code}")
print(f"Response: {response.text}")
except requests.exceptions.ConnectionError:
print("Connection failed! Make sure the HTTP bridge is running on localhost:8000")
except requests.exceptions.Timeout:
print("Request timeout! The streaming took too long")
except Exception as e:
print(f"Error: {str(e)}")
async def test_regular_chat():
"""Test regular (non-streaming) chat for comparison"""
url = "http://localhost:8000/api/chat"
payload = {
"message": "Say hello in 3 different languages",
"model": "mistral:latest",
"temperature": 0.7,
"max_tokens": 100
}
print("\nTesting regular chat endpoint for comparison...")
print(f"Sending: {payload['message']}")
print("="*60)
try:
response = requests.post(url, json=payload, timeout=30)
if response.status_code == 200:
data = response.json()
print("Regular chat successful!")
print(f"Response: {data.get('response', 'No response')}")
else:
print(f"HTTP Error: {response.status_code}")
print(f"Response: {response.text}")
except Exception as e:
print(f"Error: {str(e)}")
def test_health():
"""Test if the server is running"""
try:
response = requests.get("http://localhost:8000/health", timeout=5)
if response.status_code == 200:
print("MCP HTTP Bridge is running")
return True
else:
print(f"Health check failed: {response.status_code}")
return False
except:
print("MCP HTTP Bridge is not running")
return False
async def main():
print("MCP Streaming Chat Test")
print("="*60)
# Check if server is running
if not test_health():
print("\nTo start the server, run:")
print(" cd Z:\\Code\\MCP")
print(" python http_bridge.py")
return
# Test streaming
await test_streaming_endpoint()
# Test regular chat for comparison
await test_regular_chat()
print("\nTesting completed!")
if __name__ == "__main__":
asyncio.run(main())
