lumo-sdk 0.2.2

A Python SDK for the Lumo API by Starlight

Lumo SDK

A Python SDK for the Lumo API by Starlight. This SDK provides a simple and intuitive interface for executing tasks using AI models and tools.

Installation

pip install lumo-sdk

Quick Start

from lumo_sdk import LumoClient

# Initialize the client with your API key
client = LumoClient(api_key="your-api-key-here")

# Run a task
response = client.run_task(
    task="What is the weather in Berlin?",
    model="gpt-4.1-mini",
    base_url="https://api.openai.com/v1/chat/completions",
    tools=["ExaSearchTool", "VisitWebsite"]
)

# Access the final answer
print(response.final_answer)

# Access the steps taken
for step in response.steps:
    print(f"LLM Output: {step.llm_output}")
    print(f"Tool Calls: {step.tool_calls}")

Features

• ✅ Simple and intuitive API
• ✅ Full type hints for better IDE support
• ✅ Pydantic models for request/response validation
• ✅ Comprehensive error handling
• ✅ Support for conversation history
• ✅ Support for custom tools and agent types

API Reference

LumoClient

The main client class for interacting with the Lumo API.

Initialization

client = LumoClient(api_key="your-api-key", base_url=None)

• api_key (required): Your Lumo API key
• base_url (optional): Custom base URL (defaults to https://api.starlight-search.com)

Methods

stream_task()

Stream task execution and receive events in real-time.

for event in client.stream_task(
    task: str,
    model: str,
    base_url: str,
    tools: Optional[List[str]] = None,
    max_steps: Optional[int] = None,
    agent_type: Optional[str] = None,
    history: Optional[List[Message]] = None,
) -> Iterator[Union[StreamTokenEvent, StreamStepEvent, StreamDoneEvent]]

Parameters: Same as run_task()

Yields:

• StreamTokenEvent: Individual tokens from the LLM response (has content field)
• StreamStepEvent: Step information with tool calls and output (has step dict)
• StreamDoneEvent: Signals completion of the task

Example:

for event in client.stream_task(
    task="What is the weather?",
    model="gpt-4.1-mini",
    base_url="https://api.openai.com/v1/chat/completions",
    tools=["ExaSearchTool"]
):
    if isinstance(event, StreamTokenEvent):
        print(event.content, end="", flush=True)
    elif isinstance(event, StreamStepEvent):
        print(f"\nStep: {event.step}")

run_task()

Execute a task using the specified model and tools.

response = client.run_task(
    task: str,
    model: str,
    base_url: str,
    tools: Optional[List[str]] = None,
    max_steps: Optional[int] = None,
    agent_type: Optional[str] = None,
    history: Optional[List[Message]] = None,
) -> RunTaskResponse

Parameters:

• task (required): The task to execute
• model (required): Model ID (e.g., gpt-4, qwen2.5, gemini-2.0-flash)
• base_url (required): Base URL for the upstream API
• tools (optional): Array of tool names to make available
• max_steps (optional): Maximum number of steps to take
• agent_type (optional): Type of agent to use (function-calling or mcp)
• history (optional): Array of Message objects containing prior conversation context

Returns: RunTaskResponse object with:

• final_answer: The final answer from the task execution
• steps: List of steps taken during task execution

Streaming

The SDK also supports streaming responses for real-time token delivery:

from lumo_sdk import LumoClient, StreamTokenEvent, StreamStepEvent, StreamDoneEvent

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

for event in client.stream_task(
    task="What is the weather in Berlin?",
    model="gpt-4.1-mini",
    base_url="https://api.openai.com/v1/chat/completions",
    tools=["ExaSearchTool", "VisitWebsite"]
):
    if isinstance(event, StreamTokenEvent):
        # Print tokens as they arrive
        print(event.content, end="", flush=True)
    elif isinstance(event, StreamStepEvent):
        # Handle step with tool calls
        step = event.step
        tool_calls = step.get("tool_calls", [])
        for tool_call in tool_calls:
            print(f"\nTool: {tool_call.get('name')}")
    elif isinstance(event, StreamDoneEvent):
        print("\n✅ Done!")
        break

Examples

Basic Usage

from lumo_sdk import LumoClient

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

response = client.run_task(
    task="What is the weather in Berlin?",
    model="gpt-4.1-mini",
    base_url="https://api.openai.com/v1/chat/completions",
    tools=["ExaSearchTool", "VisitWebsite"]
)

print(response.final_answer)

With Conversation History

from lumo_sdk import LumoClient, Message, MessageRole

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

# Create conversation history
history = [
    Message(role=MessageRole.USER, content="Hello!"),
    Message(role=MessageRole.ASSISTANT, content="Hi! How can I help you?"),
]

response = client.run_task(
    task="What did I just say?",
    model="gpt-4.1-mini",
    base_url="https://api.openai.com/v1/chat/completions",
    history=history
)

print(response.final_answer)

With Max Steps

from lumo_sdk import LumoClient

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

response = client.run_task(
    task="Research the latest AI developments",
    model="gpt-4.1-mini",
    base_url="https://api.openai.com/v1/chat/completions",
    tools=["ExaSearchTool"],
    max_steps=5
)

print(f"Final answer: {response.final_answer}")
print(f"Number of steps: {len(response.steps)}")

Error Handling

from lumo_sdk import LumoClient
from lumo_sdk.exceptions import (
    LumoAuthenticationError,
    LumoRateLimitError,
    LumoServerError,
    LumoAPIError,
)

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

try:
    response = client.run_task(
        task="What is the weather?",
        model="gpt-4.1-mini",
        base_url="https://api.openai.com/v1/chat/completions"
    )
except LumoAuthenticationError:
    print("Invalid API key")
except LumoRateLimitError:
    print("Rate limit exceeded. Please try again later.")
except LumoServerError:
    print("Server error. Please try again later.")
except LumoAPIError as e:
    print(f"API error: {e}")

Models

Message

Represents a message in the conversation history.

from lumo_sdk import Message, MessageRole

message = Message(
    role=MessageRole.USER,
    content="Hello!",
    tool_call_id=None,  # Optional
    tool_calls=None  # Optional
)

MessageRole

Enum for message roles:

• MessageRole.USER
• MessageRole.ASSISTANT
• MessageRole.SYSTEM
• MessageRole.TOOL

RunTaskResponse

Response from the run_task() method.

response = client.run_task(...)
print(response.final_answer)  # str
print(response.steps)  # List[Step]

Step

Represents a step in the task execution.

for step in response.steps:
    print(step.llm_output)  # Optional[str]
    print(step.tool_calls)  # List[ToolCall]

Error Handling

The SDK provides specific exception types for different error scenarios:

• LumoError: Base exception for all SDK errors
• LumoAPIError: General API errors (400, etc.)
• LumoAuthenticationError: Authentication errors (401)
• LumoRateLimitError: Rate limit errors (429)
• LumoServerError: Server errors (500+)

Requirements

• Python 3.8+
• requests >= 2.28.0
• pydantic >= 2.0.0

Development

Setup

# Clone the repository
git clone https://github.com/starlight-search/lumo-sdk.git
cd lumo-sdk

# Install in development mode
pip install -e ".[dev]"

Running Tests

pytest

Code Formatting

black lumo_sdk tests
ruff check lumo_sdk tests

License

MIT License - see LICENSE file for details.

Links

• API Documentation
• Starlight Search

Building and Publishing to PyPI

Prerequisites

pip install build twine

Build the Package

python -m build

This will create distribution files in the dist/ directory.

Upload to PyPI

Test PyPI (for testing)

python -m twine upload --repository testpypi dist/*

Production PyPI

python -m twine upload dist/*

You'll need to have PyPI credentials configured. You can use:

• ~/.pypirc file
• Environment variables
• Interactive prompts

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.