Python SDK

from octen import Octen

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

results = client.search.search(
    query="blog post about artificial intelligence",
    count=10
)

results = client.search.search(
    query="climate tech news",
    count=20,
    start_time="2024-01-01T00:00:00Z",
    include_domains=["techcrunch.com", "wired.com"]
)

from octen import HighlightOptions, FullContentOptions

results = client.search.search(
    query="vector database comparison",
    count=10,
    highlight=HighlightOptions(enable=True, max_tokens=500),
    full_content=FullContentOptions(enable=True, max_tokens=2000),
    format="markdown"
)

results = client.search.simple_search("Python web framework", count=5)

response = client.extract.extract(
    urls=[
        "https://docs.octen.ai/api-reference/search",
        "https://www.who.int/news-room/fact-sheets/detail/influenza-(seasonal)",
    ]
)

for result in response.data.results:
    if result.status == "success":
        print(result.title, result.full_content[:200])

response = client.extract.extract(
    urls=["https://www.who.int/news-room/fact-sheets/detail/influenza-(seasonal)"],
    query="vaccination guidelines"
)

for result in response.data.results:
    if result.status == "success":
        for highlight in result.highlights:
            print(highlight)

response = client.extract.extract(
    urls=["https://example.com/article"],
    include_images=True,
    include_videos=True,
    include_favicon=True
)

for result in response.data.results:
    if result.status == "success":
        for image in result.images:
            print(image.url)

response = client.extract.extract(urls=[
    "https://docs.octen.ai/api-reference/search",
    "https://example.com/non-existent",
])

for result in response.data.results:
    if result.status == "success":
        print(f"[OK] {result.url}")
    else:
        print(f"[FAIL] {result.url}: {result.error_message}")

# Only successful URLs are billed
print(f"Billed: {response.meta.usage.successful_urls}/{response.meta.usage.total_urls}")

from octen import ChatMessage

response = client.chat.create(
    model="openai/gpt-5.4",
    messages=[ChatMessage(role="user", content="What is quantum computing?")]
)
print(response.text)

response = client.chat.create(
    model="openai/gpt-5.4",
    messages=[
        ChatMessage(role="system", content="You are a helpful assistant."),
        ChatMessage(role="user", content="What is Python?"),
        ChatMessage(role="assistant", content="Python is a programming language."),
        ChatMessage(role="user", content="What are its main use cases?"),
    ]
)

response = client.chat.create(
    model="openai/gpt-5.4",
    messages=[ChatMessage(role="user", content="Latest AI news today")],
    web_search="on"
)
print(response.text)
print(response.search_results)
print(response.citations)

from octen import WebSearchOptions

response = client.chat.create(
    model="openai/gpt-5.4",
    messages=[ChatMessage(role="user", content="Climate tech breakthroughs")],
    web_search="on",
    web_search_options=WebSearchOptions(
        count=10,
        include_domains=["nature.com", "science.org"],
        start_time="2025-01-01T00:00:00Z",
        format="markdown",
        full_content={"enable": True, "max_tokens": 2000}
    )
)

stream = client.chat.create(
    model="openai/gpt-5.4",
    messages=[ChatMessage(role="user", content="Write a short poem")],
    stream=True
)

for event in stream:
    if event.type == "content":
        print(event.choices[0].delta.content or "", end="", flush=True)
    elif event.type == "search_done":
        print(f"\nSearch results: {event.search_results}")
    elif event.type == "usage":
        print(f"\nTokens: {event.usage.total_tokens}")

response = client.embedding.create(
    input="Hello, world!",
    model="octen-embedding-4b"
)
vector = response.get_first_embedding()

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

# Embed a search query
vector = client.embedding.embed_query("search text")

# Embed a batch of documents
vectors = client.embedding.embed_documents(["doc 1", "doc 2", "doc 3"])

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 ChatMessage

async with AsyncOcten(api_key="your-api-key") as client:
    search_result, embed_result, chat_result = await asyncio.gather(
        client.search.search(query="AI news", count=5),
        client.embedding.create(input="Hello world"),
        client.chat.create(
            model="openai/gpt-5.4",
            messages=[ChatMessage(role="user", content="Hello!")]
        )
    )

async with AsyncOcten(api_key="your-api-key") as client:
    stream = await client.chat.create(
        model="openai/gpt-5.4",
        messages=[ChatMessage(role="user", content="Tell me a story")],
        stream=True
    )
    async for event in stream:
        if event.type == "content":
            print(event.choices[0].delta.content or "", end="", flush=True)

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}")