test_connection.py

#!/usr/bin/env python3
"""Test script to verify Bob's connections"""

import asyncio
import httpx
from src.config import settings


async def test_connections():
    print("Testing Bob's connections...\n")
    
    # Test MCP Server
    print("1. Testing MCP Server...")
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(f"http://localhost:{settings.mcp_server_port}/health")
            if response.status_code == 200:
                data = response.json()
                print(f"   ✓ MCP Server: {data['status']}")
                print(f"   ✓ PostgreSQL: {data['postgres']}")
                print(f"   ✓ Redis: {data['redis']}")
            else:
                print(f"   ✗ MCP Server returned status {response.status_code}")
    except Exception as e:
        print(f"   ✗ MCP Server connection failed: {e}")
    
    print("\n2. Testing Memory Tools...")
    try:
        async with httpx.AsyncClient() as client:
            # Create a test KB
            response = await client.post(
                f"http://localhost:{settings.mcp_server_port}/tools/kb/create/test_kb"
            )
            if response.status_code == 200:
                print("   ✓ Created test knowledge base")
            
            # Store a test entity
            response = await client.post(
                f"http://localhost:{settings.mcp_server_port}/tools/kb/store",
                json={
                    "kb_name": "test_kb",
                    "content": "Hello, I am Bob!",
                    "metadata": {"test": True}
                }
            )
            if response.status_code == 200:
                print("   ✓ Stored test entity")
            
            # Search test
            response = await client.post(
                f"http://localhost:{settings.mcp_server_port}/tools/kb/search",
                json={
                    "kb_name": "test_kb",
                    "query": "Bob"
                }
            )
            if response.status_code == 200:
                results = response.json()
                print(f"   ✓ Search returned {len(results)} results")
            
            # Test STM
            response = await client.post(
                f"http://localhost:{settings.mcp_server_port}/tools/stm/push",
                json={
                    "thread_id": "test_thread",
                    "role": "user",
                    "content": "Test message"
                }
            )
            if response.status_code == 200:
                print("   ✓ Pushed message to STM")
            
    except Exception as e:
        print(f"   ✗ Memory tools test failed: {e}")
    
    print("\nAll tests complete!")


if __name__ == "__main__":
    asyncio.run(test_connections())