refactor: remove curl examples and improve HTTP API examples organization

- Remove curl_examples.sh as authentication is too complex for simple curl commands
- Move complex examples to advanced/ subfolder for better organization
- Keep simple_working_example.py prominent as the recommended starting point
- Add comprehensive CI test for HTTP examples
- Update documentation to highlight the working example and learning path
- Improve user experience by focusing on Python examples with automatic auth
This commit is contained in:
LearningCircuit
2025-11-01 01:19:44 +01:00
parent 01b6f163a5
commit bdcb934cbe
7 changed files with 350 additions and 87 deletions

View File

@@ -0,0 +1,67 @@
# Advanced HTTP API Examples
This folder contains more comprehensive HTTP API examples for learning advanced features and production use cases.
## 📁 Available Examples
### 📚 `simple_http_example.py` - **COMPREHENSIVE GUIDE**
-**Automatic user creation**
- 📊 **Multiple API examples** (research, settings, history)
- 🔍 **Progress monitoring** with status updates
- ⏱️ **Runtime:** 3-15 minutes (more comprehensive testing)
**Perfect for:** Learning different API endpoints, understanding the full API surface
### 🚀 `http_api_examples.py` - **ADVANCED CLIENT**
- 🔧 **Reusable client class** for integration
- 📈 **Advanced features** (batch processing, polling)
- 🎛️ **Comprehensive patterns** for production use
- ⏱️ **Runtime:** 5-30 minutes (extensive testing)
**Perfect for:** Building applications, production integration, advanced use cases
## 🚀 Quick Start
### 1. Run the Comprehensive Example
```bash
cd advanced
python simple_http_example.py
```
### 2. Try the Advanced Client
```bash
cd advanced
python http_api_examples.py
```
## ⚠️ Important Notes
- **Longer Runtime**: These examples take longer than the basic example
- **More Features**: They demonstrate additional API endpoints and patterns
- **Learning Focused**: Designed to help you understand the full API surface
- **Production Ready**: Advanced examples include patterns for production use
## 📚 What These Examples Demonstrate
### simple_http_example.py
- ✅ All basic functionality from the main example
- ✅ Settings management (get/update configuration)
- ✅ Research history access
- ✅ Progress polling and monitoring
- ✅ Multiple research scenarios
- ✅ Error handling patterns
### http_api_examples.py
- ✅ All functionality from simple_http_example.py
- ✅ Reusable client class for application integration
- ✅ Batch processing capabilities
- ✅ Advanced polling strategies
- ✅ Production-ready error handling
- ✅ Comprehensive API coverage
- ✅ Settings management patterns
## 🔗 Related Documentation
- [Main HTTP Examples](../README.md)
- [Basic Working Example](../simple_working_example.py)
- [API Documentation](../../../README.md)

View File

@@ -0,0 +1,484 @@
#!/usr/bin/env python3
"""
HTTP API Examples for Local Deep Research v1.0+
This script demonstrates comprehensive usage of the LDR HTTP API with authentication.
Includes examples for research, settings management, and batch operations.
Requirements:
- LDR v1.0+ (with authentication features)
- LDR server running: python -m local_deep_research.web.app
- Beautiful Soup: pip install beautifulsoup4
This example works COMPLETELY out of the box - no manual setup required!
"""
import time
from typing import Any, Dict, List
import requests
from pathlib import Path
import sys
from bs4 import BeautifulSoup
# Add the src directory to Python path for programmatic user creation
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src"))
from local_deep_research.database.encrypted_db import DatabaseManager
from local_deep_research.database.models import User
from local_deep_research.database.auth_db import get_auth_db_session
# Configuration
BASE_URL = "http://localhost:5000"
def create_test_user():
"""Create a test user programmatically - works out of the box!"""
username = f"testuser_{int(time.time())}"
password = "testpassword123"
print(f"Creating test user: {username}")
try:
# Create user in auth database
auth_db = get_auth_db_session()
new_user = User(username=username)
auth_db.add(new_user)
auth_db.commit()
auth_db.close()
# Create encrypted database for user
db_manager = DatabaseManager()
db_manager.create_user_database(username, password)
print(f"✅ User created successfully: {username}")
return username, password
except Exception as e:
print(f"❌ Failed to create user: {e}")
return None, None
class LDRClient:
"""Client for interacting with LDR API v1.0+ with authentication"""
def __init__(self, base_url: str = BASE_URL):
self.base_url = base_url
self.session = requests.Session()
self.csrf_token = None
self.username = None
def login(self, username: str, password: str) -> bool:
"""Authenticate with the LDR server."""
try:
# Get login page and CSRF token
login_page = self.session.get(f"{self.base_url}/auth/login")
if login_page.status_code != 200:
return False
soup = BeautifulSoup(login_page.text, "html.parser")
csrf_input = soup.find("input", {"name": "csrf_token"})
login_csrf = csrf_input.get("value")
if not login_csrf:
return False
# Submit login form
login_response = self.session.post(
f"{self.base_url}/auth/login",
data={
"username": username,
"password": password,
"csrf_token": login_csrf,
},
allow_redirects=False,
)
if login_response.status_code not in [200, 302]:
return False
self.username = username
# Get API CSRF token for API calls
csrf_response = self.session.get(f"{self.base_url}/auth/csrf-token")
if csrf_response.status_code == 200:
self.csrf_token = csrf_response.json().get("csrf_token")
return True
except Exception:
return False
def logout(self) -> None:
"""Logout from the server."""
if self.csrf_token:
self.session.post(
f"{self.base_url}/auth/logout",
headers={"X-CSRF-Token": self.csrf_token},
)
def _get_headers(self) -> Dict[str, str]:
"""Get headers with CSRF token."""
return {"X-CSRF-Token": self.csrf_token} if self.csrf_token else {}
def check_health(self) -> Dict[str, Any]:
"""Check API health status."""
response = self.session.get(f"{self.base_url}/auth/check")
return response.json()
def start_research(self, query: str, **kwargs) -> Dict[str, Any]:
"""Start a new research task."""
payload = {
"query": query,
"model": kwargs.get("model"),
"search_engines": kwargs.get("search_engines", ["wikipedia"]),
"iterations": kwargs.get("iterations", 2),
"questions_per_iteration": kwargs.get("questions_per_iteration", 3),
"temperature": kwargs.get("temperature", 0.7),
"local_context": kwargs.get("local_context", 2000),
"web_context": kwargs.get("web_context", 2000),
}
response = self.session.post(
f"{self.base_url}/research/api/start",
json=payload,
headers=self._get_headers(),
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"Failed to start research: {response.text}")
def get_research_status(self, research_id: str) -> Dict[str, Any]:
"""Get the status of a research task."""
response = self.session.get(
f"{self.base_url}/research/api/research/{research_id}/status"
)
return response.json()
def get_research_result(self, research_id: str) -> Dict[str, Any]:
"""Get the results of a completed research task."""
response = self.session.get(
f"{self.base_url}/research/api/research/{research_id}/result"
)
return response.json()
def wait_for_research(
self, research_id: str, timeout: int = 300
) -> Dict[str, Any]:
"""Wait for research to complete and return results."""
start_time = time.time()
while time.time() - start_time < timeout:
status = self.get_research_status(research_id)
if status.get("status") == "completed":
return self.get_research_result(research_id)
elif status.get("status") == "failed":
raise Exception(
f"Research failed: {status.get('error', 'Unknown error')}"
)
print(
f" Status: {status.get('status', 'unknown')} - {status.get('progress', 'N/A')}"
)
time.sleep(3)
raise TimeoutError(
f"Research {research_id} timed out after {timeout} seconds"
)
def get_settings(self) -> Dict[str, Any]:
"""Get all user settings."""
response = self.session.get(f"{self.base_url}/settings/api")
return response.json()
def get_setting(self, key: str) -> Any:
"""Get a specific setting value."""
response = self.session.get(f"{self.base_url}/settings/api/{key}")
if response.status_code == 200:
return response.json()
return None
def update_setting(self, key: str, value: Any) -> bool:
"""Update a setting value."""
response = self.session.put(
f"{self.base_url}/settings/api/{key}",
json={"value": value},
headers=self._get_headers(),
)
return response.status_code in [200, 201]
def get_history(self, limit: int = 10) -> List[Dict[str, Any]]:
"""Get research history."""
response = self.session.get(
f"{self.base_url}/history/api", params={"limit": limit}
)
data = response.json()
return data.get("items", data.get("history", []))
def get_available_models(self) -> Dict[str, str]:
"""Get available LLM providers and models."""
response = self.session.get(
f"{self.base_url}/settings/api/available-models"
)
data = response.json()
return data.get("providers", data.get("models", {}))
def get_available_search_engines(self) -> List[str]:
"""Get available search engines."""
response = self.session.get(
f"{self.base_url}/settings/api/available-search-engines"
)
data = response.json()
return data.get("engines", data.get("engine_options", []))
def example_quick_research(client: LDRClient) -> None:
"""Example: Quick research with minimal parameters."""
print("\n=== Example 1: Quick Research ===")
research = client.start_research(
query="What are the key principles of machine learning?",
iterations=1,
questions_per_iteration=2,
)
print(f"Started research ID: {research['research_id']}")
# Wait for completion
result = client.wait_for_research(research["research_id"])
print(f"\nSummary: {result['summary'][:500]}...")
print(f"Sources: {len(result.get('sources', []))}")
print(f"Findings: {len(result.get('findings', []))}")
def example_detailed_research(client: LDRClient) -> None:
"""Example: Detailed research with multiple search engines."""
print("\n=== Example 2: Detailed Research ===")
# Check available search engines
engines = client.get_available_search_engines()
print(f"Available search engines: {engines}")
# Use multiple engines
selected_engines = (
["wikipedia", "arxiv"] if "arxiv" in engines else ["wikipedia"]
)
research = client.start_research(
query="Impact of climate change on global food security",
search_engines=selected_engines,
iterations=3,
questions_per_iteration=4,
temperature=0.7,
)
print(f"Started detailed research ID: {research['research_id']}")
# Monitor progress
result = client.wait_for_research(research["research_id"], timeout=600)
print(f"\nTitle: {result.get('query', 'N/A')}")
print(f"Summary length: {len(result['summary'])} characters")
print(f"Sources: {len(result.get('sources', []))}")
# Show some findings
findings = result.get("findings", [])
if findings:
print("\nTop findings:")
for i, finding in enumerate(findings[:3], 1):
print(f"{i}. {finding.get('text', 'N/A')[:100]}...")
def example_settings_management(client: LDRClient) -> None:
"""Example: Managing user settings."""
print("\n=== Example 3: Settings Management ===")
# Get current settings
settings = client.get_settings()
settings_data = settings.get("settings", {})
# Display current LLM configuration
llm_provider = settings_data.get("llm.provider", {}).get("value", "Not set")
llm_model = settings_data.get("llm.model", {}).get("value", "Not set")
print(f"Current LLM Provider: {llm_provider}")
print(f"Current LLM Model: {llm_model}")
# Get available models
models = client.get_available_models()
print(f"\nAvailable providers: {list(models.keys())}")
# Example: Update temperature setting
current_temp = settings_data.get("llm.temperature", {}).get("value", 0.7)
print(f"\nCurrent temperature: {current_temp}")
# Update temperature (example - uncomment to actually update)
# success = client.update_setting("llm.temperature", 0.5)
# print(f"Temperature update: {'Success' if success else 'Failed'}")
def example_batch_research(client: LDRClient) -> None:
"""Example: Running multiple research tasks in batch."""
print("\n=== Example 4: Batch Research ===")
queries = [
"What is quantum entanglement?",
"How does CRISPR gene editing work?",
"What are the applications of blockchain technology?",
]
research_ids = []
# Start all research tasks
for query in queries:
try:
research = client.start_research(
query=query, iterations=1, questions_per_iteration=2
)
research_ids.append(
{
"id": research["research_id"],
"query": query,
"status": "started",
}
)
print(f"Started: {query} (ID: {research['research_id']})")
except Exception as e:
print(f"Failed to start '{query}': {e}")
# Wait for all to complete
print("\nWaiting for batch completion...")
completed = 0
while completed < len(research_ids):
for research in research_ids:
if research["status"] != "completed":
try:
status = client.get_research_status(research["id"])
if status.get("status") == "completed":
research["status"] = "completed"
completed += 1
print(f"✓ Completed: {research['query']}")
except Exception:
pass
if completed < len(research_ids):
time.sleep(3)
# Get all results
print("\nBatch Results Summary:")
for research in research_ids:
try:
result = client.get_research_result(research["id"])
print(f"\n{research['query']}:")
print(f" - Summary: {result['summary'][:150]}...")
print(f" - Sources: {len(result.get('sources', []))}")
except Exception as e:
print(f" - Error getting results: {e}")
def example_research_history(client: LDRClient) -> None:
"""Example: Viewing research history."""
print("\n=== Example 5: Research History ===")
history = client.get_history(limit=5)
if not history:
print("No research history found.")
return
print(f"Found {len(history)} recent research items:\n")
for item in history:
created = item.get("created_at", "Unknown date")
query = item.get("query", "Unknown query")
status = item.get("status", "Unknown")
research_id = item.get("id", item.get("research_id", "N/A"))
print(f"ID: {research_id}")
print(f"Query: {query}")
print(f"Date: {created}")
print(f"Status: {status}")
print("-" * 40)
def main():
"""Run all examples."""
print("=== LDR HTTP API v1.0 Examples ===")
print(
"🎯 This example works completely out of the box - no manual setup required!"
)
# First, check if server is running
try:
response = requests.get(f"{BASE_URL}/auth/check", timeout=5)
if response.status_code in [
200,
401,
]: # 401 is expected when not authenticated
print("✅ Server is running")
else:
print(f"❌ Server returned status code: {response.status_code}")
print("Make sure the server is running:")
print(" python -m local_deep_research.web.app")
return
except requests.exceptions.ConnectionError:
print("❌ Cannot connect to LDR server!")
print("Make sure the server is running:")
print(" python -m local_deep_research.web.app")
return
# Create test user automatically
username, password = create_test_user()
if not username:
print("❌ Failed to create test user")
return
# Create client
client = LDRClient(BASE_URL)
try:
# Login with the created user
print(f"\nLogging in as: {username}")
if not client.login(username, password):
print("❌ Login failed!")
return
print("✅ Login successful")
# Check health
health = client.check_health()
print(f"Authenticated: {health.get('authenticated', False)}")
print(f"Username: {health.get('username', 'N/A')}")
# Run examples
example_quick_research(client)
example_detailed_research(client)
example_settings_management(client)
example_batch_research(client)
example_research_history(client)
print("\n✅ All examples completed successfully!")
print(f"🔑 Created user: {username}")
print("📝 You can now use this user for manual testing:")
print(f" Username: {username}")
print(f" Password: {password}")
print(f" Login URL: {BASE_URL}/auth/login")
except requests.exceptions.ConnectionError:
print("\n❌ Cannot connect to LDR server!")
print("Make sure the server is running:")
print(" python -m local_deep_research.web.app")
except Exception as e:
print(f"\n❌ Error: {e}")
finally:
# Always logout
client.logout()
print("\n✅ Logged out")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,418 @@
#!/usr/bin/env python3
"""
Simple HTTP API Example for Local Deep Research v1.0+
This example shows how to use the LDR API with authentication.
Works completely out of the box with automatic user creation.
"""
import requests
import time
import sys
from bs4 import BeautifulSoup
from pathlib import Path
# Add the src directory to Python path for programmatic user creation
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src"))
from local_deep_research.database.encrypted_db import DatabaseManager
from local_deep_research.database.models import User
from local_deep_research.database.auth_db import get_auth_db_session
# Configuration
API_URL = "http://localhost:5000"
def create_test_user():
"""Create a test user programmatically."""
username = f"testuser_{int(time.time())}"
password = "testpassword123"
print(f"Creating test user: {username}")
try:
# Create user in auth database
auth_db = get_auth_db_session()
new_user = User(username=username)
auth_db.add(new_user)
auth_db.commit()
auth_db.close()
# Create encrypted database for user
db_manager = DatabaseManager()
db_manager.create_user_database(username, password)
print(f"✅ User created successfully: {username}")
return username, password
except Exception as e:
print(f"❌ Failed to create user: {e}")
return None, None
def main():
print("=== LDR HTTP API Example ===")
print("🎯 This example works completely out of the box!\n")
print("⚠️ IMPORTANT NOTES:")
print(" • This script may take several minutes to complete")
print(" • Research progress can be monitored in the server logs")
print(" • Server logs are available at: /tmp/ldr_server.log")
print(
" • Use 'tail -f /tmp/ldr_server.log' to monitor progress in real-time"
)
print(" • Results will be available at the URL shown when complete\n")
# Check if server is running
try:
response = requests.get(f"{API_URL}/", timeout=5)
if response.status_code != 200:
print("❌ Server is not responding correctly")
print("\n📋 HOW TO START THE SERVER:")
print(" • Option 1: python -m local_deep_research.web.app")
print(
" • Option 2: bash scripts/dev/restart_server.sh (recommended)"
)
print(
" • Note: restart_server.sh will kill existing server process"
)
sys.exit(1)
print("✅ Server is running")
except Exception:
print(
"❌ Cannot connect to server. Please make sure it's running on http://localhost:5000"
)
print("\n📋 HOW TO START THE SERVER:")
print(" • Option 1: python -m local_deep_research.web.app")
print(" • Option 2: bash scripts/dev/restart_server.sh (recommended)")
print(" • Note: restart_server.sh will kill existing server process")
sys.exit(1)
# Create test user automatically
username, password = create_test_user()
if not username:
print("❌ Failed to create test user")
sys.exit(1)
# Create a session to persist cookies
session = requests.Session()
print(f"\nTesting with user: {username}")
# Step 1: Login
print("\n1. Authenticating...")
# Get login page and CSRF token
login_page = session.get(f"{API_URL}/auth/login")
soup = BeautifulSoup(login_page.text, "html.parser")
csrf_input = soup.find("input", {"name": "csrf_token"})
login_csrf = csrf_input.get("value")
if not login_csrf:
print("❌ Could not get CSRF token from login page")
sys.exit(1)
# Login with form data (not JSON)
login_response = session.post(
f"{API_URL}/auth/login",
data={
"username": username,
"password": password,
"csrf_token": login_csrf,
},
allow_redirects=False,
)
if login_response.status_code not in [200, 302]:
print(f"❌ Login failed: {login_response.text}")
print("\nPlease ensure:")
print("- The server is running: python -m local_deep_research.web.app")
sys.exit(1)
print("✅ Login successful")
# Step 2: Get CSRF token
print("\n2. Getting CSRF token...")
csrf_response = session.get(f"{API_URL}/auth/csrf-token")
csrf_token = csrf_response.json()["csrf_token"]
headers = {"X-CSRF-Token": csrf_token}
print("✅ CSRF token obtained")
# Initialize research_id to None
research_id = None
# Example 1: Quick Summary (using the start endpoint)
print("\n=== Example 1: Quick Summary ===")
print(
"📝 This example demonstrates starting a research query and polling for results"
)
print("⏱️ This typically takes 1-3 minutes to complete\n")
research_request = {
"query": "What is machine learning?",
"model": None, # Will use default from settings
"search_engines": ["wikipedia"], # Fast for demo
"iterations": 1,
"questions_per_iteration": 2,
}
# Start research - CORRECT ENDPOINT
print("🚀 Starting research...")
start_response = session.post(
f"{API_URL}/api/start_research", json=research_request, headers=headers
)
if start_response.status_code != 200:
print(f"❌ Failed to start research: {start_response.text}")
sys.exit(1)
research_data = start_response.json()
research_id = research_data["research_id"]
print("✅ Research started successfully!")
print(f"🆔 Research ID: {research_id}")
print("📊 Monitor progress in server logs: tail -f /tmp/ldr_server.log")
print(f"🌐 Results will be available at: {API_URL}/results/{research_id}\n")
# Poll for results
print("⏳ Waiting for research to complete...")
print(
"⚠️ NOTE: This will poll for up to 3 minutes to ensure research completes"
)
print(
" If it fails, the research may still be running - check the results URL\n"
)
poll_count = 0
max_polls = 18 # Maximum 3 minutes (18 * 10 seconds)
while poll_count < max_polls:
status_response = session.get(
f"{API_URL}/api/research/{research_id}/status"
)
if status_response.status_code == 200:
status = status_response.json()
current_status = status.get("status", "unknown")
progress = status.get("progress", 0)
poll_count += 1
elapsed_time = poll_count * 10 # 10 seconds per poll
print(
f" Check {poll_count} ({elapsed_time}s): Status = {current_status} (Progress: {progress}%)"
)
if current_status == "completed":
print("🎉 Research completed successfully!")
break
elif current_status == "failed":
print(
f"❌ Research failed: {status.get('error', 'Unknown error')}"
)
print(
"📋 Check server logs for details: tail -f /tmp/ldr_server.log"
)
sys.exit(1)
elif current_status in ["queued", "in_progress"]:
# Continue polling
pass
else:
print(f"⚠️ Unexpected status: {current_status}")
else:
print(
f"⚠️ Status check failed with code: {status_response.status_code}"
)
time.sleep(10) # Wait 10 seconds between polls
if poll_count >= max_polls:
print("⏰ 3-minute timeout reached - research is still running")
print("💡 This is normal for complex research queries!")
print(f"📊 Check results later at: {API_URL}/results/{research_id}")
print("📋 Monitor progress with: tail -f /tmp/ldr_server.log")
print(
"🔍 The script will still try to fetch results (may be incomplete)"
)
# Get results
results_response = session.get(
f"{API_URL}/research/api/research/{research_id}/result"
)
if results_response.status_code == 200:
results = results_response.json()
print(f"\n📝 Summary: {results['summary'][:300]}...")
print(f"📚 Sources: {len(results.get('sources', []))} found")
print(f"🔍 Findings: {len(results.get('findings', []))} findings")
# Example 2: Check Settings
print("\n=== Example 2: Current Settings ===")
settings_response = session.get(f"{API_URL}/settings/api")
if settings_response.status_code == 200:
settings = settings_response.json()["settings"]
# Show some key settings
llm_provider = settings.get("llm.provider", {}).get("value", "Not set")
llm_model = settings.get("llm.model", {}).get("value", "Not set")
print(f"LLM Provider: {llm_provider}")
print(f"LLM Model: {llm_model}")
# Example 3: Get Research History
print("\n=== Example 3: Research History ===")
history_response = session.get(f"{API_URL}/history/api")
if history_response.status_code == 200:
history = history_response.json()
items = history.get("items", history.get("history", []))
print(f"Found {len(items)} research items")
for item in items[:3]: # Show first 3
print(
f"- {item.get('query', 'Unknown query')} ({item.get('created_at', 'Unknown date')})"
)
# Example 4: Get and Display Research Results (with retry logic)
print("\n=== Example 4: Research Results ===")
if research_id:
print(f"📄 Fetching research results for ID: {research_id}")
print(
"🔄 Will retry until results are available (up to 2 additional minutes)\n"
)
# Retry fetching results until available
results_retries = 0
max_results_retries = 12 # 2 minutes (12 * 10 seconds)
while results_retries < max_results_retries:
results_response = session.get(
f"{API_URL}/api/research/{research_id}/report"
)
if results_response.status_code == 200:
# Results are available, parse and display them
results = results_response.json()
content = results.get("content", "")
sources = results.get("sources", [])
findings = results.get("findings", [])
print(
f"✅ Results retrieved successfully after {(results_retries + 1) * 10} seconds!"
)
print("\n📝 RESEARCH SUMMARY:")
print("=" * 50)
if content:
# Show first 500 characters of the summary
summary_preview = (
content[:500] + "..." if len(content) > 500 else content
)
print(summary_preview)
else:
print("No summary content available")
print(f"\n📚 SOURCES FOUND: {len(sources)}")
for i, source in enumerate(
sources[:3], 1
): # Show first 3 sources
title = source.get("title", "Unknown Title")
url = source.get("url", "No URL")
print(f" {i}. {title}")
print(f" {url}")
if len(sources) > 3:
print(f" ... and {len(sources) - 3} more sources")
print(f"\n🔍 KEY FINDINGS: {len(findings)}")
for i, finding in enumerate(
findings[:3], 1
): # Show first 3 findings
finding_text = finding.get("text", "No finding text")
finding_preview = (
finding_text[:150] + "..."
if len(finding_text) > 150
else finding_text
)
print(f" {i}. {finding_preview}")
if len(findings) > 3:
print(f" ... and {len(findings) - 3} more findings")
print(
f"\n🌐 View full results at: {API_URL}/results/{research_id}"
)
print("=" * 50)
print("🎉 Results displayed successfully!")
break # Exit retry loop - success!
elif results_response.status_code == 404:
results_retries += 1
elapsed_time = results_retries * 10
print(
f" Retry {results_retries}/{max_results_retries} ({elapsed_time}s): Results not ready yet, waiting..."
)
time.sleep(10) # Wait 10 seconds before retrying
else:
print(
f"❌ Failed to fetch results: {results_response.status_code}"
)
print(f"Response: {results_response.text[:200]}")
break # Exit retry loop - error
# Handle case where max retries reached
if results_retries >= max_results_retries:
print(
f"\n⏰ Maximum retry time reached ({max_results_retries * 10} seconds)"
)
print("💡 This is normal for complex research queries!")
print(f"📊 Check results later at: {API_URL}/results/{research_id}")
print("📋 Monitor progress with: tail -f /tmp/ldr_server.log")
print(
"🔍 The research is still running - results will be available when complete"
)
else:
print(
"⚠️ No research ID available - research may not have started properly"
)
# Logout
print("\n5. Logging out...")
session.post(f"{API_URL}/auth/logout", headers=headers)
print("✅ Logged out successfully")
if __name__ == "__main__":
print("🎯 Simple LDR HTTP API Example - Works out of the box!")
print("⚡ This script creates a user automatically and tests the API")
print(
"⏱️ Total runtime: Up to 3 minutes polling + 2 minutes results retry + research time"
)
print(
"🔄 Automatically retries fetching results until available (up to 2 minutes)\n"
)
print("📋 REQUIREMENTS:")
print(" • LDR server running")
print(" • Beautiful Soup: pip install beautifulsoup4\n")
print("🚀 START THE SERVER:")
print(" • Option 1: python -m local_deep_research.web.app")
print(" • Option 2: bash scripts/dev/restart_server.sh (recommended)")
print(" • Note: restart_server.sh will kill existing server process\n")
print("📊 MONITORING:")
print(" • Server logs: tail -f /tmp/ldr_server.log")
print(" • This script polls for up to 3 minutes")
print(" • If research takes longer, script shows where to check results\n")
print("⏰ TIMING INFO:")
print(" • Script polls for 3 minutes to let research complete")
print(" • Then retries fetching results for up to 2 additional minutes")
print(" • Research typically completes in 2-10 minutes")
print(" • Script displays results automatically when available")
print(
" • If timeout reached, results URL provided for checking completion\n"
)
main()