API Reference

Streaming Search

⚠️ Important Note: The auto-generated code samples in the playground will not work for this WebSocket endpoint, with the exception of the Python and JS(WebSocket) examples shown. Please use these provided examples only, and not the auto-generated ones.

This endpoint provides a WebSocket connection for streaming search results in real-time.

Connection

Connect to: wss://api.critique-labs.ai/v1/ws/search

Headers

X-API-Key: (Required) Your unique API key for authentication.

Message Format

After connecting, send a JSON message with the following structure:

{
    "prompt": "your search query",
    "image": "optional base64 image or URL",
    "source_blacklist": ["optional list of domains to exclude"],
    "output_format": {
        // Optional structured output format
    }
}

Streaming Responses

The server will stream responses as JSON messages with the following structure:

{
    "type": "response" | "context" | "error",
    "content": "the actual content"
}

type: "response" - Contains generated response content
type: "context" - Contains source context information
type: "error" - Contains error messages if any occur

GET

import websockets
import asyncio
import json

async def search_streaming():
    uri = "wss://api.critique-labs.ai/v1/ws/search"
    headers = {
        'X-API-Key': '<YOUR API KEY HERE>'
    }
    
    async with websockets.connect(uri, extra_headers=headers) as websocket:
        # Send the search request
        await websocket.send(json.dumps({
            'prompt': 'how much are flights to this place from Hong Kong right now',
            'source_blacklist': ['expedia.com'],  # optional - specify either source_blacklist or source_whitelist, not both
            # 'source_whitelist': ['kayak.com'],  # optional - specify either source_blacklist or source_whitelist, not both
            'output_format': {
                'flights': [{"origin": "string", "airline": "string", "destination": "string", "price": "number"}],
                "response": "string"
            }
        }))
        
        # Receive and process streaming responses
        while True:
            try:
                response = await websocket.recv()
                data = json.loads(response)
                
                # Handle different types of responses
                if data['type'] == 'response':
                    print('Response:', data['content'])
                elif data['type'] == 'context':
                    print('Context:', data['content'])
                elif data['type'] == 'error':
                    print('Error:', data['content'])
                    break
            except websockets.exceptions.ConnectionClosed:
                print("Connection closed")
                break

# Run the async function
asyncio.run(search_streaming())

Headers

X-API-Key

string

required

Your API key for authentication

Body

application/json

prompt

string

required

The search query, can be as extended or succinct as you like

image

string

Optional, This string can be the url of the image you want to send. Alternatively, (and the preferred method) is providing a base64 encoded string of the image. If providing a url, the link must start with https:// and ends with a common image file extension like .jpg, .jpeg, .png, .gif, etc.

source_blacklist

string[]

Optional, a list of strings representing domains you want to exclude from the agentic search. ['cnn.com','foxnews.com'].

output_format

object

Optional, a json schema for the response. This will be used to format the response of the agentic search. The types allowed can be any of: string, number, boolean, integer, array

Response

101

WebSocket connection established

Delete Api With Key

import websockets
import asyncio
import json

async def search_streaming():
    uri = "wss://api.critique-labs.ai/v1/ws/search"
    headers = {
        'X-API-Key': '<YOUR API KEY HERE>'
    }
    
    async with websockets.connect(uri, extra_headers=headers) as websocket:
        # Send the search request
        await websocket.send(json.dumps({
            'prompt': 'how much are flights to this place from Hong Kong right now',
            'source_blacklist': ['expedia.com'],  # optional - specify either source_blacklist or source_whitelist, not both
            # 'source_whitelist': ['kayak.com'],  # optional - specify either source_blacklist or source_whitelist, not both
            'output_format': {
                'flights': [{"origin": "string", "airline": "string", "destination": "string", "price": "number"}],
                "response": "string"
            }
        }))
        
        # Receive and process streaming responses
        while True:
            try:
                response = await websocket.recv()
                data = json.loads(response)
                
                # Handle different types of responses
                if data['type'] == 'response':
                    print('Response:', data['content'])
                elif data['type'] == 'context':
                    print('Context:', data['content'])
                elif data['type'] == 'error':
                    print('Error:', data['content'])
                    break
            except websockets.exceptions.ConnectionClosed:
                print("Connection closed")
                break

# Run the async function
asyncio.run(search_streaming())