gl-odr 0.0.1b6

GL Deep Research Python Client - Python client library for GL DeepResearch API

GL Deep Research

A Python library for interacting with the GL DeepResearch API, providing deep research capabilities via asynchronous tasks and task groups.

Prerequisites

• Python >=3.11,<3.13

Installation

pip install gl-odr

Or with uv:

uv add gl-odr

Quick Start

from gl_odr import DeepResearchClient

# Initialize the client
client = DeepResearchClient(api_key="your-api-key")

# Health check
health = client.health.check()
print(f"Status: {health.status}")

# Create async task
task = client.tasks.create(
    query="Research topic",
    profile="GPTR-QUICK"
)
print(f"Task ID: {task.task_id}")

# Check task status
status = client.tasks.get_status(task.task_id)
print(f"Status: {status.status}")

# Stream task events in real time
for event in client.tasks.stream(task.task_id):
    print(f"{event.event_type}: {event.value}")

# Get task result when complete
result = client.tasks.get(task.task_id)
if result.data:
    print(result.data.result)

Configuration

Environment Variables

The library supports the following environment variables:

• GLODR_API_KEY: API key for authentication
• GLODR_BASE_URL: Base URL for the API (default: https://stag-gl-deep-research.obrol.id/)

Client Initialization

from gl_odr import DeepResearchClient

# Using explicit parameters
client = DeepResearchClient(
    api_key="your-api-key",
    base_url="https://stag-gl-deep-research.obrol.id/",
    timeout=300.0,
    default_headers={"X-Custom-Header": "value"}
)

# Using environment variables
import os
os.environ["GLODR_API_KEY"] = "your-api-key"
os.environ["GLODR_BASE_URL"] = "gldr-base-url"
client = DeepResearchClient()

Health Check

health = client.health.check()
print(f"Status: {health.status}")  # "ok"

Tasks API

Create Async Task

from gl_odr import WebhookConfig

# Create task without webhook
task = client.tasks.create(
    query="Research the history of AI",
    profile="GPTR-QUICK"
)
print(f"Task ID: {task.task_id}")

# Create task with webhook notification
task = client.tasks.create(
    query="Research the history of AI",
    profile="GPTR-QUICK",
    webhook=WebhookConfig(
        url="https://your-server.com/webhook",
        secret="your-secret"
    )
)

Get Task Result and Status

# Get full task result
result = client.tasks.get(task.task_id)
print(f"Status: {result.status}")
if result.data:
    print(f"Result: {result.data.result}")

# Get lightweight status only
status = client.tasks.get_status(task.task_id)
print(f"Status: {status.status}")

Stream Task Events

# Stream task events (available for 24 hours)
for event in client.tasks.stream(task.task_id):
    print(f"{event.event_type}: {event.value}")

Taskgroups API

Create Taskgroup

from gl_odr import WebhookConfig

# Create taskgroup with multiple queries
taskgroup = client.taskgroups.create(
    profile="GPTR-QUICK",
    query=[
        "What is quantum computing?",
        "What is machine learning?",
        "What is blockchain?"
    ],
    webhook=WebhookConfig(
        url="https://your-server.com/webhook",
        secret="your-secret"
    )
)
print(f"Taskgroup ID: {taskgroup.taskgroup_id}")
print(f"Tasks: {taskgroup.tasks}")

# Create empty taskgroup (add tasks later)
taskgroup = client.taskgroups.create(profile="GPTR-QUICK")

Manage Taskgroup

# Get taskgroup status
group = client.taskgroups.get(taskgroup.taskgroup_id)
print(f"Status: {group.status}")  # EMPTY, PENDING, STARTED, SUCCESS, FAILURE, PARTIAL_FAILURE
print(f"Tasks: {group.tasks}")

# Add task to existing group
new_task = client.taskgroups.add_task(
    taskgroup_id=taskgroup.taskgroup_id,
    query="What is cloud computing?"
)
print(f"New Task ID: {new_task.task_id}")

# Get specific task from group
task_result = client.taskgroups.get_task(
    taskgroup_id=taskgroup.taskgroup_id,
    task_id=new_task.task_id
)
print(f"Task Status: {task_result.status}")

Stream Taskgroup Events

# Stream all tasks in group
for event in client.taskgroups.stream(taskgroup.taskgroup_id):
    print(f"{event.event_type}: {event.value}")

# Stream specific task in group
for event in client.taskgroups.stream_task(
    taskgroup_id=taskgroup.taskgroup_id,
    task_id=task.task_id
):
    print(f"{event.event_type}: {event.value}")

Webhook Management

Create Task or Taskgroup with Webhook

from gl_odr import WebhookConfig

# Task with webhook
task = client.tasks.create(
    query="Research topic",
    profile="GPTR-QUICK",
    webhook=WebhookConfig(url="https://your-server.com/webhook", secret="your-secret")
)

# Taskgroup with webhook
taskgroup = client.taskgroups.create(
    profile="GPTR-QUICK",
    query=["Query 1", "Query 2"],
    webhook=WebhookConfig(url="https://your-server.com/webhook", secret="your-secret")
)

Verify Webhook Signature

import hashlib
import hmac

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    """Verify webhook signature using HMAC-SHA256."""
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

# In your webhook handler:
# signature = request.headers.get("X-Webhook-Signature")
# if verify_webhook_signature(request.body, signature, "your-secret"):
#     # Process webhook

Error Handling

import httpx
from gl_odr import DeepResearchClient

client = DeepResearchClient(api_key="your-api-key")

try:
    task = client.tasks.create(
        query="Research topic",
        profile="INVALID-PROFILE"
    )
except httpx.HTTPStatusError as e:
    print(f"HTTP Error: {e.response.status_code}")
    print(f"Response: {e.response.text}")
except ValueError as e:
    print(f"Validation Error: {e}")

Examples

For more detailed examples, see the examples directory:

Example	Description
01_async_tasks.py	Async task creation with polling
02_taskgroups.py	Batch research with taskgroups
03_streaming.py	Stream task events in real time
04_webhooks.py	Webhook configuration and verification

Development

Install Development Dependencies

uv sync --group dev

Running Unit Tests

The library uses pytest for unit testing. All test files are located in the tests/ directory.

Run All Tests

uv run pytest

Run Tests with Coverage Report

uv run pytest --cov=gl_odr --cov-report=term-missing

This will show:

• Test coverage percentage
• Which lines are covered/missing
• Coverage report in the terminal

Run Specific Test Files

# Run only client tests
uv run pytest tests/test_client.py

# Run only health tests
uv run pytest tests/test_health.py

# Run only tasks tests
uv run pytest tests/test_tasks.py

# Run only taskgroups tests
uv run pytest tests/test_taskgroups.py

Run Specific Test Functions

# Run a specific test function
uv run pytest tests/test_client.py::test_client_initialization_with_api_key

# Run tests matching a pattern
uv run pytest -k "test_client"

Run Tests with Verbose Output

# Show detailed output for each test
uv run pytest -v

# Show even more details including print statements
uv run pytest -v -s

Other Useful Options

# Stop on first failure
uv run pytest -x

# Show local variables on failure
uv run pytest -l

# Run tests matching a keyword
uv run pytest -k "tasks"

Run Linting

uv run ruff check .
uv run ruff format .

API Reference

• GL DeepResearch API Documentation
• Health API
• Tasks API
• Taskgroups API

License

MIT License - see LICENSE file for details.