Python SDK

from octen import Octen

client = Octen(api_key="your-api-key") # or set OCTEN_API_KEY env var

response = client.search.broad_search(
    query="compare cloud GPU pricing across providers",
    max_queries=5,  # up to N sub-queries (1-30, default 5)
)
# response.queries -> sub-queries; response.search_results -> grouped results

results = client.search.search(
    query="blog post about artificial intelligence",
    count=10,
    time_range="week",
    include_domains=["techcrunch.com", "wired.com"],
)

response = client.search.news_search(
    query="EU AI Act enforcement",
    time_range="week",
    count=10,
)
# response.subjects -> storylines; response.results -> ranked articles

response = client.search.business_search(
    query="compare Stripe and Adyen payment processing",
    entities={"count": 2, "max_news": 5},
)
# response.entities -> entity cards; response.results -> ranked sources

response = client.place_search.search(
    "coffee shops",
    latitude=35.6812,
    longitude=139.7671,
    radius=1000,  # meters (1-50000), needs both coordinates
    places={"count": 5},
)
# response.places -> place cards, each with a distance

response = client.search.pulse_search("github docs", count=5)
# response.results -> titles and URLs; lite mode returns URLs without a scheme

results = client.image_search.search(
    query="golden retriever puppy",
    count=5,
)

results = client.video_search.search(
    query="how to make fresh pasta",
    count=5,
)

response = client.extract.extract(
    urls=["https://docs.octen.ai/api-reference/search", "https://octen.ai"],
    format="markdown",
)

for item in response.items:
    if item.status == "success":
        print(item.title, item.full_content[:200])
    else:
        print(f"[FAIL] {item.url}: {item.error_message}")

from octen import ChatMessage, OctenSearchTool

response = client.chat.create(
    model="openai/gpt-5.4",
    messages=[ChatMessage(role="user", content="What happened in tech today?")],
    tools=[OctenSearchTool()],  # enable live web search
)
print(response.text)
print(response.search_results)

response = client.embedding.create(
    input=["first document", "second document"],
    model="octen-embedding-8b",
    input_type="document",
)
vectors = response.get_embeddings()

response = client.vl_embedding.create(
    model="octen-vl-embedding-large",
    contents=[
        {"text": "A cute orange cat on a wooden chair"},
        {"image": "https://example.com/cat.jpg"},
    ],
    enable_fusion=True,
)
vector = response.get_first_embedding()

import asyncio
from octen import AsyncOcten

async def main():
    async with AsyncOcten(api_key="your-api-key") as client:
        results = await client.search.search(query="machine learning startups", count=10)

asyncio.run(main())

from octen import (
    OctenAuthenticationError,
    OctenRateLimitError,
    OctenTimeoutError,
    OctenConnectionError,
    OctenStreamError,
    OctenAPIError,
)

try:
    results = client.search.search("query")
except OctenAuthenticationError:
    print("Invalid API key")
except OctenRateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")
except OctenTimeoutError:
    print("Request timed out")
except OctenConnectionError:
    print("Network connection failed")
except OctenStreamError as e:
    print(f"Stream error: {e.message} (code: {e.code})")
except OctenAPIError as e:
    print(f"API error: {e.status_code} - {e.message}")