Octen Python SDK

Official Python SDK for the Octen API — web search, URL extraction, text embeddings, and multi-model LLM chat in one package.

✨ Features

• 🔍 Web Search — search and retrieve ranked web results with filtering, highlighting, and full content
• 🌐 URL Extract — fetch and parse 1-20 URLs in a single batch with markdown / text output, standard / advanced / auto modes, query-driven highlights, page links, and media (images / videos / audio / favicon)
• 💬 Multi-model Chat — access 15+ LLMs (GPT, Claude, Gemini, Kimi, MiniMax, Qwen, DeepSeek) through a single unified API
• 🧮 Text Embeddings — convert text into high-quality vector representations
• 🖼️ VL (Multimodal) Embeddings — encode text, images, and videos into a single fused vector or independent per-element vectors
• ⚡ Streaming (SSE) — real-time token streaming with typed event objects
• 🔄 Auto Retry — exponential backoff for transient errors
• 🛡️ Type Safe — full Pydantic models with IDE auto-completion
• 🔀 Async Support — native asyncio client for concurrent workloads
• 📦 HTTP/2 — connection pooling and keep-alive out of the box

📦 Installation

pip install octen

Requires Python 3.8 or higher.

Development Version

pip install octen[dev]

Async Support

pip install octen[async]

🚀 Quick Start

Search

from octen import Octen

with Octen(api_key="your-api-key") as client:
    response = client.search.search(query="Python programming", count=5)

    for result in response.results:
        print(f"Title: {result['title']}")
        print(f"URL: {result['url']}")
        print(f"Highlight: {result.get('highlight', '')}")

Broad Search

Decompose a query into multiple sub-queries, search them concurrently, and get results grouped per sub-query. Best for comparisons (pricing, products, vendors), surveys/research, and multi-angle questions where a single search() only reaches a few subtopics — pass the query as-is and raise max_queries for broader coverage.

from octen import Octen

with Octen(api_key="your-api-key") as client:
    response = client.search.broad_search(
        query="compare cloud GPU pricing across providers",
        max_queries=5,   # up to N sub-queries (1-30, default 5)
        count=10,        # per-sub-query SearchOptions, same as search()
    )

    print("Sub-queries:", response.queries)
    for group in response.search_results:
        print(f"\n# {group['query']}  ({group.get('latency')} ms)")
        for result in group["results"]:
            print(f"  {result['title']} — {result['url']}")

Extract

from octen import Octen

with Octen(api_key="your-api-key") as client:
    response = client.extract.extract(
        urls=["https://example.com", "https://octen.ai"],
        format="markdown",
    )

    for item in response.items:
        if item.status == "success":
            print(f"{item.title} — {item.url}")
        else:
            print(f"FAILED {item.url}: {item.error_message}")

Chat

from octen import Octen, ChatMessage, OctenSearchTool

with Octen(api_key="your-api-key") as client:
    response = client.chat.create(
        model="openai/gpt-5.4",
        messages=[ChatMessage(role="user", content="Hello!")],
        tools=[OctenSearchTool()],  # enable live web search
    )
    print(response.text)

Embeddings

from octen import Octen

with Octen(api_key="your-api-key") as client:
    embedding = client.embedding.create(
        input=["Hello, world!"],
        model="octen-embedding-4b"
    )
    vector = embedding.get_first_embedding()
    print(f"Vector dimension: {len(vector)}")

VL (Multimodal) Embeddings

from octen import Octen

with Octen(api_key="your-api-key") as client:
    # Fuse text + image into one vector
    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()
    if vector is not None:
        print(f"Fused vector dimension: {len(vector)}")

🔍 Search API

Advanced Search

from octen import Octen, HighlightOptions, FullContentOptions

with Octen(api_key="your-api-key") as client:
    response = client.search.search(
        query="machine learning best practices",
        count=10,
        topic="news",  # general (default) or news
        include_domains=["github.com", "arxiv.org"],  # Search only these domains
        language=["en", "zh"],  # Filter results by language (ISO 639-1 codes)
        time_range="month",  # Relative time window
        include_images=True,  # Return image URLs per result
        highlight=HighlightOptions(
            enable=True,
            max_tokens=500
        ),
        full_content=FullContentOptions(
            enable=True,
            max_tokens=2000
        ),
        timeout=60.0  # Custom timeout
    )

    print(f"Found {len(response.results)} results")
    print(f"Token usage: {response.usage}")

🌐 Extract API

Fetch and parse one or more URLs into structured content. A single request accepts 1-20 URLs and is served as one upstream batch.

Batch Extract

from octen import Octen

with Octen(api_key="your-api-key") as client:
    response = client.extract.extract(
        urls=[
            "https://example.com",
            "https://octen.ai",
            "https://www.iana.org/about",
        ],
        format="markdown",          # "text" or "markdown" (default markdown)
        max_age_seconds=600,        # accept results cached within 10 minutes
        timeout=30,                 # per-URL upstream fetch budget (1-60s)
        include_images=True,        # also returns cover_image when the page has one
        request_timeout=90.0,       # local httpx deadline; >= timeout + overhead
    )

    print(f"OK: {response.successful_urls}/{response.total_urls}  ({response.latency}ms)")

    for item in response.items:
        if item.status == "success":
            print(f"  ✓ {item.title} — {item.url}")
            if item.favicon:  # favicon is returned by default when available
                print(f"    favicon: {item.favicon}")
            if item.cover_image:
                print(f"    cover: {item.cover_image.url}")
        else:
            # Partial-success is first-class: failed URLs don't poison siblings.
            print(f"  ✗ {item.url}: {item.error_message}")

Note: response.items is parsed lazily and raises OctenAPIError if any row fails to parse (signals server schema drift). The raw dicts remain accessible via response.results as an escape hatch.

Choosing a Mode

mode selects how each URL is fetched. Leave it unset and the server uses "standard"; the SDK never injects a default.

Scenario	mode
News / blogs / docs / product pages (ordinary sites)	standard (or omit)
Known hard sites (Reddit / X / LinkedIn / ResearchGate / anti-bot / SPA)	advanced
Mixed batch, unsure which are hard	auto
Cost-sensitive, mostly ordinary sites	standard
Quality first, mostly hard sites	advanced

• standard — fastest and cheapest ($1 / 1K successful URLs). On login-walled, JS-heavy or anti-bot sites it may return empty content or only a page skeleton and still report status="success".
• advanced — highest success rate: real-browser rendering with stronger anti-bot handling. Slower, and 2.5x the price ($2.5 / 1K).
• auto — picks per URL, escalating to advanced only for hard URLs. Each URL is billed at the mode it actually used, so a batch of mostly hard sites costs close to all-advanced.

Advanced and auto are slower: raise timeout (e.g. to 60) and request_timeout with it. Failed URLs are not billed.

from octen import Octen

with Octen(api_key="your-api-key") as client:
    response = client.extract.extract(
        urls=["https://example.com", "https://www.reddit.com/r/Python/"],
        mode="auto",
        timeout=60,
        request_timeout=90.0,
    )

    for item in response.items:
        # The mode actually used; auto resolves per URL, and an advanced
        # request can resolve to standard. Absent on failed items.
        print(item.url, item.status, item.resolved_mode)

    # Billing is authoritative from meta.usage.successful_by_mode.
    print(response.standard_urls, response.advanced_urls)

Standard first, advanced retry. For batches that are mostly ordinary sites, run standard and retry only the misses with advanced. A miss is a failed item, or a "success" that came back empty or skeletal:

def needs_retry(item):
    if item.status == "failed":
        return True
    return bool(item.page_structure) and item.page_structure.primary == "No Main Content"

response = client.extract.extract(urls=urls)
retry = [item.url for item in response.items if needs_retry(item)]
if retry:
    retried = client.extract.extract(urls=retry, mode="advanced", timeout=60, request_timeout=90.0)

See examples/extract_basic.py for the full version.

Page Links

include_links returns each page's links on item.links, to discover a site's pages or follow outbound references. Pass a dict or an ExtractLinksOptions; {} uses the defaults.

response = client.extract.extract(
    urls=["https://octen.ai"],
    include_links={"scope": "prefer_external", "max_links": 50},
)
for link in response.items[0].links or []:
    print(link.is_external, link.anchor_text, link.url)

• scope — "prefer_internal" (default) lists same-site links first; "prefer_external" lists links to other registered domains first. Page order is kept within each group.
• max_links — 1-1000, default 200. Out-of-range values raise OctenValidationError before the request is sent (the server rejects them rather than clamping).

Query-driven Highlights

from octen import Octen

with Octen(api_key="your-api-key") as client:
    response = client.extract.extract(
        urls=["https://en.wikipedia.org/wiki/Python_(programming_language)"],
        query="async programming",   # max 500 chars
    )

    for item in response.items:
        for snippet in item.highlights or []:
            print(f"• {snippet}")

Single URL (Convenience Shortcut)

from octen import Octen

with Octen(api_key="your-api-key") as client:
    response = client.extract.simple_extract("https://example.com")
    print(response.items[0].title)

Two Timeouts, Two Layers

timeout and request_timeout operate at different layers — easy to confuse, important to get right:

Parameter	Layer	Controls	On timeout
timeout (int, 1-60s)	server	upstream per-URL fetch budget	the slow URL is reported with status="failed" and an error_message; sibling URLs in the same batch are returned normally as long as the upstream responds within the bounded round trip
request_timeout (float)	client (httpx)	local socket deadline for the whole HTTP call	raises OctenTimeoutError

Rule of thumb: request_timeout >= timeout + network_overhead.

🖼️ Image Search API

In Beta — contact us to request beta access.

Multimodal image search by text query and/or a reference image URL.

from octen import Octen

with Octen(api_key="your-api-key") as client:
    response = client.image_search.search(
        query="golden retriever puppy",   # text input
        # image_url="https://example.com/ref.jpg",  # optional reference image
        topic="general",                  # "general" or "design"
        count=5,                          # 1-10 (default 5)
    )

    for result in response.results:
        print(result["title"], result["url"], result.get("thumbnail"))

🎬 Video Search API

In Beta — contact us to request beta access.

Text-driven video search.

from octen import Octen

with Octen(api_key="your-api-key") as client:
    response = client.video_search.search(
        query="how to make fresh pasta",
        count=5,                          # 1-10 (default 5)
    )

    for result in response.results:
        print(result["title"], result["url"], result.get("duration_seconds"))

💬 Chat API

Web search is now a tool. The old top-level web_search / web_search_options parameters were removed. Enable web search by passing an OctenSearchTool in tools; tune it with OctenSearchToolParameters. For broad (multi-query) search, pass an OctenBroadSearchTool with OctenBroadSearchToolParameters instead.

Non-streaming

from octen import Octen, ChatMessage, OctenSearchTool, OctenSearchToolParameters

with Octen(api_key="your-api-key") as client:
    response = client.chat.create(
        model="openai/gpt-5.4",
        messages=[
            ChatMessage(role="system", content="You are a helpful assistant."),
            ChatMessage(role="user", content="What happened in tech today?"),
        ],
        tools=[OctenSearchTool(parameters=OctenSearchToolParameters(safesearch="off", count=5))],
        max_tokens=500,
        temperature=0.7
    )

    print(response.text)
    print(f"Tokens used: {response.usage.total_tokens}")

    # Access search results
    if response.search_results:
        for group in response.search_results:
            for item in group.results:
                print(f"  - {item.title}: {item.url}")

Streaming

from octen import Octen, ChatMessage, OctenSearchTool

with Octen(api_key="your-api-key") as client:
    for event in client.chat.create(
        model="openai/gpt-5.4",
        messages=[ChatMessage(role="user", content="Tell me a story")],
        stream=True,
        tools=[OctenSearchTool()],
    ):
        if event.type == "search_done":
            print(f"[{len(event.search_results or [])} search groups]")

        elif event.type == "content" and event.choices:
            print(event.choices[0].delta.content or "", end="", flush=True)

        elif event.type == "finish":
            print()  # newline

        elif event.type == "usage" and event.usage:
            print(f"[total tokens: {event.usage.total_tokens}]")

Tool Calling

from octen import Octen
from octen.models import ChatMessage, Tool, ToolFunction

weather_tool = Tool(
    function=ToolFunction(
        name="get_weather",
        description="Get current weather for a city",
        parameters={
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "City name"},
            },
            "required": ["city"],
        }
    )
)

with Octen(api_key="your-api-key") as client:
    response = client.chat.create(
        model="openai/gpt-5.4",
        messages=[ChatMessage(role="user", content="What's the weather in London?")],
        tools=[weather_tool],
        tool_choice="auto"
    )
    if response.choices[0].finish_reason == "tool_calls":
        tc = response.choices[0].message.tool_calls[0]
        print(f"Tool: {tc.function.name}, Args: {tc.function.arguments}")

JSON Output Mode

from octen import Octen, ChatMessage
from octen.models import ResponseFormat

with Octen(api_key="your-api-key") as client:
    response = client.chat.create(
        model="google/gemini-3-flash-preview",
        messages=[ChatMessage(role="user", content="Return a JSON list of 3 programming languages")],
        response_format=ResponseFormat(type="json_object"),
    )
    print(response.text)

Web Search with Full Page Content

from octen import Octen, ChatMessage, OctenSearchTool, OctenSearchToolParameters
from octen.models.search import FullContentOptions

with Octen(api_key="your-api-key") as client:
    response = client.chat.create(
        model="openai/gpt-5.4",
        messages=[ChatMessage(role="user", content="Latest Python 3.13 features?")],
        tools=[OctenSearchTool(parameters=OctenSearchToolParameters(
            safesearch="off",
            full_content=FullContentOptions(enable=True, max_tokens=2048),
        ))],
    )
    print(f"Full content tokens: {response.usage.full_content_tokens}")

Broad Web Search (multi-query)

The octen_broad_search tool fans the question out into up to max_queries sub-queries and searches them concurrently — a peer of octen_search in tools. Results land in response.search_results as usual.

from octen import Octen, ChatMessage, OctenBroadSearchTool, OctenBroadSearchToolParameters

with Octen(api_key="your-api-key") as client:
    response = client.chat.create(
        model="openai/gpt-5.4",
        messages=[ChatMessage(role="user", content="Compare the leading vector databases")],
        tools=[OctenBroadSearchTool(parameters=OctenBroadSearchToolParameters(
            max_queries=3,   # up to 3 generated sub-queries (1-30, default 5)
            count=5,         # results per sub-query
        ))],
    )
    print(response.text)
    for group in response.search_results or []:
        print(f"[{group.query}] {len(group.results)} results")

Prompt Caching (content blocks)

from octen import Octen, ChatMessage, ChatContentBlock, CacheControl

with Octen(api_key="your-api-key") as client:
    response = client.chat.create(
        model="anthropic/claude-sonnet-4.6",
        messages=[
            ChatMessage(role="user", content=[
                ChatContentBlock(
                    type="text",
                    text="<long reusable context...>",
                    cache_control=CacheControl(ttl="1h"),  # 5m (default) or 1h
                ),
                ChatContentBlock(type="text", text="Summarise the above."),
            ]),
        ],
    )
    print(response.usage.prompt_tokens_details.cached_tokens)

🤖 Supported Chat Models

For the full and up-to-date list of supported models, visit the Octen official website.

🧮 Embeddings API

Batch Embeddings

from octen import Octen

with Octen(api_key="your-api-key") as client:
    # Process multiple texts
    texts = [
        "Artificial intelligence is transforming the world",
        "Applications of deep learning",
        "Natural language processing technology"
    ]

    response = client.embedding.create(
        input=texts,
        model="octen-embedding-8b",
        input_type="document"
    )

    vectors = response.get_embeddings()
    print(f"Generated {len(vectors)} vectors")

    # Or use convenience methods
    query_vector = client.embedding.embed_query("search query")
    doc_vectors = client.embedding.embed_documents(["document 1", "document 2"])

Multimodal (VL) Embeddings

The vl_embedding resource encodes text, images, and videos through the /vl-embedding endpoint. Two output modes:

• Fusion (enable_fusion=True) — all contents collapse into a single vector (type="fusion")
• Independent (enable_fusion omitted or False) — one vector per content element (type="vl"). Omitting the field inherits the server default (False).

Default-value semantics: enable_fusion / fps / instruct default to None in the SDK and are stripped from the wire via exclude_none=True. When omitted, the server applies its own defaults (shown in the table). enable_fusion=False and omitting the field produce different wire payloads but currently the same behavior — pass False explicitly only if you need to override a future server default change.

Parameter	Type	Required	Description
model	string	✓	octen-vl-embedding (max dim 2048) or octen-vl-embedding-large (max dim 4096)
contents	list[dict | VLEmbeddingContent]	✓	Each item has exactly one of text / image / video (non-empty). Max 20 total, 5 images, 1 video
enable_fusion	bool	—	Single fused vector vs. independent vectors. Omit → server default False
dimension	int	—	Output dim, ≤ model's max. Omit → model's max
fps	float (0-1)	—	Video frame sampling density. Omit → server default 1; lower values reduce video token cost
instruct	string	—	Custom task description. Omit → server default "Represent the user's input."

Content element constraints:

• text — string, ≤ 32,000 tokens per entry
• image — URL or Base64 string, ≤ 5MB, formats: JPEG/PNG/WEBP/BMP/TIFF/ICO/DIB/ICNS/SGI
• video — URL only, ≤ 50MB, formats: MP4/AVI/MOV

from octen import Octen, VLEmbeddingContent

with Octen(api_key="your-api-key") as client:
    # 1) Text-only embedding
    response = client.vl_embedding.create(
        model="octen-vl-embedding",
        contents=[{"text": "What is multimodal vector search?"}],
    )
    vector = response.get_first_embedding()
    if vector is not None:
        print(f"text-only dim={len(vector)}")  # 2048

    # 2) Fused multimodal embedding (text + images + video)
    response = client.vl_embedding.create(
        model="octen-vl-embedding-large",
        contents=[
            {"text": "Outdoor tent, 3-4 person, waterproof"},
            {"image": "https://example.com/tent_setup.jpg"},
            {"image": "https://example.com/tent_inside.jpg"},
            {"video": "https://example.com/tent_demo.mp4"},
        ],
        enable_fusion=True,
        dimension=2048,
        fps=0.3,
        instruct="Represent the outdoor product for retrieval",
    )
    print(f"fusion type={response.items[0].type} dim={response.items[0].dimension}")

    # 3) Independent per-image embeddings
    response = client.vl_embedding.create(
        model="octen-vl-embedding",
        contents=[
            {"image": "https://example.com/product_1.jpg"},
            {"image": "https://example.com/product_2.jpg"},
        ],
        enable_fusion=False,
    )
    for item in response.items:
        print(f"index={item.index} type={item.type} dim={item.dimension}")

    # 4) Typed content objects (alternative to dicts)
    response = client.vl_embedding.create(
        model="octen-vl-embedding",
        contents=[
            VLEmbeddingContent(text="A photo of a cat"),
            VLEmbeddingContent(image="https://example.com/cat.jpg"),
        ],
        enable_fusion=True,
    )

    # Usage breakdown
    if response.usage:
        print(
            f"input_tokens={response.usage.input_tokens} "
            f"text_tokens={response.usage.text_tokens} "
            f"image_tokens={response.usage.image_tokens} "
            f"image_count={response.usage.image_count} "
            f"duration={response.usage.duration}s"
        )

Async usage:

import asyncio
from octen import AsyncOcten

async def main():
    async with AsyncOcten(api_key="your-api-key") as client:
        response = await client.vl_embedding.create(
            model="octen-vl-embedding-large",
            contents=[
                {"text": "modern art exhibition"},
                {"image": "https://example.com/painting.jpg"},
            ],
            enable_fusion=True,
        )
        print(response.get_first_embedding()[:3])

asyncio.run(main())

📖 Full API reference: https://docs.octen.ai/api-reference/vl-embedding

Custom Configuration

from octen import Octen

client = Octen(
    api_key="your-api-key",
    base_url="https://api.octen.ai",  # Custom API endpoint
    timeout=10.0,  # Global default timeout (seconds)
    max_retries=3,  # Maximum retry attempts
    http2=True  # Enable HTTP/2
)

try:
    # This request uses global timeout (10 seconds)
    response1 = client.search.search("query 1")

    # This request overrides timeout to 30 seconds
    response2 = client.search.search("complex query", timeout=30.0)
finally:
    client.close()  # Release connection pool resources

📚 API Documentation

Search API

client.search.search()

Perform a web search query.

Parameters:

• query (str, required): Search query string, max 500 characters
• topic (str, optional): Search category, options: "general" (default), "news"
• count (int, optional): Number of results to return, range 1-100, default 5
• include_domains (List[str], optional): Include only results from these domains (max 1000, each ≤30 chars)
• exclude_domains (List[str], optional): Exclude results from these domains (max 150, each ≤30 chars)
• include_text (List[str], optional): Results must contain these texts (max 5, each ≤30 chars)
• exclude_text (List[str], optional): Results must exclude these texts (max 5, each ≤30 chars)
• language (List[str], optional): Languages to filter results by, as a list of ISO 639-1 codes (e.g. ["en", "ja"]). Supported: ar, de, en, es, fr, hi, id, it, ja, ko, nl, pl, pt, ru, th, tr, vi, zh. Empty/omitted = no language filter. See https://localizely.com/iso-639-1-list/
• time_basis (str, optional): Time basis, options: "auto", "published", "crawled"
• time_range (str, optional): Relative time window, options: "day", "week", "month", "year" (or "d", "w", "m", "y")
• start_time (str, optional): Start time in ISO 8601 format
• end_time (str, optional): End time in ISO 8601 format
• highlight (HighlightOptions, optional): Highlight options (max_tokens 100–20000, default 512)
• format (str, optional): Content format, options: "text", "markdown"
• safesearch (str, optional): Safe search, options: "off", "strict" (default)
• full_content (FullContentOptions, optional): Full content options (max_tokens 100–100000, default 2048)
• include_images (bool, optional): Include image URLs in results, default false
• timeout (float, optional): Request timeout in seconds

Returns: SearchResponse object

Response Properties:

• results - List of search results
• query - The actual query used
• usage - Token usage information
• latency - Latency information

client.search.broad_search()

Decompose a query into related sub-queries, search them concurrently, and return results grouped per sub-query.

Parameters:

• query (str, required): Original search query, max 500 characters
• max_queries (int, optional): Upper bound on sub-queries generated, range 1-30, default 5
• all per-sub-query options from search() (topic, count, include_domains, exclude_domains, include_text, exclude_text, language, time_basis, time_range, start_time, end_time, highlight, format, safesearch, full_content, include_images) — applied to every sub-query (sent nested inside search_options)
• timeout (float, optional): Request timeout in seconds

Returns: BroadSearchResponse object

Response Properties:

• queries - List of generated sub-queries
• search_results - Per-sub-query result groups ({query, results, latency})
• query - The original query
• usage - Token usage information
• latency - Latency information

Extract API

client.extract.extract()

Fetch and parse one or more URLs in a single batch.

Parameters:

• urls (List[str], required): URLs to extract — 1-20 per request, each ≤ 2048 characters
• format (str, optional): Content format — "text" or "markdown" (server default "markdown")
• max_age_seconds (int, optional): Cache window in seconds (minimum 300); older cached versions are re-fetched. Defaults to 86_400 (24h) when omitted. Values outside the allowed range are adjusted server-side
• query (str, optional): Query for per-result highlight extraction, max 500 characters. When set, each item's highlights field is populated
• timeout (int, optional): Per-URL upstream fetch budget in seconds, range 1-60, default 30. A URL that exceeds this returns status="failed" — siblings continue
• include_images (bool, optional): Return image objects, and cover_image when the page has one (default False)
• include_videos (bool, optional): Return video objects (default False)
• include_audio (bool, optional): Return audio objects (default False)
• request_timeout (float, optional): Local HTTP socket deadline (httpx), distinct from the upstream timeout above. Set >= timeout + network_overhead (e.g. 90 with timeout=60)
• mode (str, optional): "standard", "advanced" or "auto". Omitted means the server uses "standard". See Choosing a Mode; with advanced/auto, raise timeout (e.g. 60)
• include_links (ExtractLinksOptions | dict, optional): Return page links. scope ("prefer_internal" default / "prefer_external"), max_links (1-1000, default 200); {} uses the defaults

Caller typos (e.g. url= instead of urls=, inculde_images=) are rejected at construction time by Pydantic — they will not silently reach the server.

Returns: ExtractResponse object

Response Properties:

• code (int) — 0 on success
• msg (str) — Server message
• request_id (str | None) — Server-generated request id; echoed from X-Request-Id if you supplied that header
• results (List[dict]) — Raw per-URL result dicts
• items (List[ExtractItem]) — Parsed, typed per-URL results
• usage (dict | None) — {"total_urls": int, "successful_urls": int, "successful_by_mode": {...}}
• total_urls (int | None) — Convenience accessor
• successful_urls (int | None) — Number of URLs with status="success" in this batch
• successful_by_mode (dict | None) — {"standard_urls": int, "advanced_urls": int}; billing is authoritative from this, not from summing resolved_mode
• standard_urls / advanced_urls (int | None) — Convenience accessors for successful_by_mode
• latency (int | None) — End-to-end server latency in ms
• warning (str | None) — Non-fatal warning

ExtractItem Fields:

• url (str) — The requested URL
• status (Literal["success", "failed"]) — Extraction outcome
• title (str | None) — Page title
• full_content (str | None) — Extracted page content
• highlights (List[str] | None) — Snippets when query was set
• time_published / time_last_crawled (str | None) — ISO 8601 timestamps
• error_message (str | None) — Populated only when status="failed"
• favicon (str | None) — Page favicon URL; returned by default when available
• cover_image (ExtractCoverImage | None) — Page cover image {url} (when include_images=True and the page has one)
• images / videos / audio (List[Any] | None) — Media objects, each {url}; kept as raw dicts so unknown fields survive
• resolved_mode (str | None) — Mode actually used for this URL: "standard" or "advanced". Always present on success, absent on failure. May differ from the requested mode
• links (List[ExtractLink] | None) — {url, anchor_text, is_external}, when include_links was set and links were found
• category (ExtractCategory | None) — primary / secondary classification labels (when classifier enabled server-side)
• page_structure (ExtractPageStructure | None) — primary / secondary structure labels

client.extract.simple_extract(url)

Convenience shortcut for a single URL with default parameters.

response = client.extract.simple_extract("https://example.com")

Chat API

client.chat.create()

Create a chat completion (non-streaming or streaming).

Parameters:

• messages (List[ChatMessage | dict], required): Conversation history. Each item can be a ChatMessage object or a plain dict {"role": ..., "content": ...}
• model (str, required): Model ID (e.g. "openai/gpt-5.4"). See Supported Chat Models for the full list
• stream (bool, optional): If True, return a Stream iterator of StreamEvent objects. Default False
• max_tokens (int, optional): Maximum number of output tokens
• max_completion_tokens (int, optional): Alternative max-token parameter
• temperature (float, optional): Sampling temperature [0, 2]
• top_p (float, optional): Nucleus sampling probability (0, 1]
• top_k (int, optional): Top-k sampling
• min_p (float, optional): Minimum probability cutoff [0, 1]
• top_a (float, optional): Top-a sampling [0, 1]
• repetition_penalty (float, optional): Repetition penalty (0, 2]
• frequency_penalty (float, optional): Frequency penalty [-2, 2]
• presence_penalty (float, optional): Presence penalty [-2, 2]
• response_format (ResponseFormat, optional): Output format — ResponseFormat(type="text"), ResponseFormat(type="json_object"), or ResponseFormat(type="json_schema", json_schema=...)
• stop (List[str], optional): Up to 4 stop sequences
• seed (int, optional): Integer seed for deterministic sampling
• reasoning_effort (str, optional): Chain-of-thought effort: "xhigh", "high", "medium", "low", "minimal", or "none"
• reasoning_max_tokens (int, optional): Max reasoning token budget (min 1024)
• verbosity (str, optional): Output verbosity: "low", "medium", or "high"
• parallel_tool_calls (bool, optional): Allow parallel tool calls (default True server-side)
• previous_response_id (str, optional): Continue from a previous response
• logprobs (bool, optional): Whether to return log probabilities
• top_logprobs (int, optional): Number of most-likely tokens [0, 20]. Requires logprobs=True
• logit_bias (Dict[str, float], optional): Token ID to bias value mapping
• tools (List[Tool | OctenSearchTool | OctenBroadSearchTool | dict], optional): Tool definitions available to the model. Three kinds:
  • Function tool — Tool(function=ToolFunction(name=..., description=..., parameters=..., strict=...))
  • Built-in web search — OctenSearchTool(parameters=OctenSearchToolParameters(...)). Parameters: max_searches (default 5) plus the shared web-search options: topic (general/news), count (1-100), include_domains/exclude_domains, include_text/exclude_text (≤5 each), language (List[str] of ISO 639-1 codes, e.g. ["en", "ja"]), time_basis (auto/published/crawled), time_range (day/week/month/year or d/w/m/y), start_time/end_time (ISO 8601), highlight (HighlightOptions), format (markdown/text), safesearch (off/strict), full_content (FullContentOptions), include_images (bool)
  • Built-in broad web search — OctenBroadSearchTool(parameters=OctenBroadSearchToolParameters(...)). Fans out into multiple sub-queries. Parameters: max_queries (1-30, default 5) plus the same shared web-search options as octen_search
• tool_choice (str | dict, optional): "none", "auto", "required", or a dict specifying a particular tool (e.g. {"type": "octen_search"} or {"type": "function", "function": {"name": ...}})
• user (str, optional): Opaque end-user identifier
• timeout (float, optional): Per-request timeout in seconds (default 60s for chat)

Returns:

• ChatCompletion when stream=False
• Stream (iterable of StreamEvent) when stream=True

ChatCompletion Properties:

• id - Unique completion ID
• model - Model used for generation
• choices - List of Choice objects. Each message exposes content, reasoning, reasoning_details (List[ReasoningDetail]), annotations (List[Annotation] with url_citation), tool_calls, and refusal
• text - Convenience accessor for the first choice's content
• reasoning - Convenience accessor for the first choice's reasoning (falls back to legacy reasoning_content)
• usage - Usage object: prompt_tokens, completion_tokens, total_tokens, num_search_queries, full_content_tokens, prompt_tokens_details (cached_tokens, cache_write_tokens), completion_tokens_details (reasoning_tokens)
• search_results - List of ChatSearchResult (when the octen_search tool ran)
• citations - Citation string referencing search results
• warning - Optional warning message

StreamEvent Properties:

• type - Event type: "search_done", "content", "finish", "usage", "error"
• choices - List of StreamChoice (with delta.content and delta.reasoning for incremental text)
• search_results - Web search results (on search_done event)
• usage - Token usage (on usage event)
• citations - Citation string (on search_done event)
• error - StreamError with message and code (on error event)

Embedding API

client.embedding.create()

Create text embedding vectors.

Parameters:

• input (str | List[str], required): Input text or list of texts
• model (str, optional): Model name, options:
  • "octen-embedding-0.6b" - Lightweight model
  • "octen-embedding-4b" - Balanced performance
  • "octen-embedding-8b" - Highest quality
• dimension (int, optional): Vector dimension
• input_type (str, optional): Input type, options: "query" or "document"
• truncation (bool, optional): Whether to truncate long inputs, default True
• timeout (float, optional): Request timeout in seconds

Returns: EmbeddingResponse object

Response Methods:

• get_embeddings() - Get all vectors
• get_first_embedding() - Get first vector (for single input)

Convenience Methods:

• embed_query(text) - Embed a single query text
• embed_documents(texts) - Batch embed document texts

VL Embedding API

client.vl_embedding.create()

Create multimodal embedding vectors from text, images, and videos.

Parameters:

• model (str, required): Model name, options:

  • "octen-vl-embedding" - Max dimension 2048
  • "octen-vl-embedding-large" - Max dimension 4096

• contents (List[dict | VLEmbeddingContent], required): Content elements. Each item must set exactly one of:

  • text (str) - Text input, ≤ 32k tokens
  • image (str) - Image URL or Base64, ≤ 5MB (JPEG/PNG/WEBP/BMP/TIFF/ICO/DIB/ICNS/SGI)
  • video (str) - Video URL, ≤ 50MB (MP4/AVI/MOV)

  Limits: ≤ 20 total elements, ≤ 5 images, ≤ 1 video per request.

• enable_fusion (bool, optional): Fuse all contents into a single vector. Omit → server default False (independent vectors). The SDK uses exclude_none=True, so omitting and passing False produce different wire payloads

• dimension (int, optional): Output dimension, ≤ model max. Omit → model max

• fps (float, optional): Video frame sampling density, 0 ≤ x ≤ 1. Omit → server default 1

• instruct (str, optional): Custom task description for the encoder. Omit → server default "Represent the user's input."

• timeout (float, optional): Request timeout in seconds

Returns: VLEmbeddingResponse object

Response Properties:

• code (int) - Response code; 0 means success. Non-zero responses raise OctenAPIError at parse time
• msg (str) - Server-provided result message
• request_id - Server-assigned request id (use this to grep server logs)
• items - List of VLEmbeddingResult with .index, .embedding, .type ("vl" or "fusion"), .dimension. Raises OctenAPIError on schema drift
• results - Raw result dicts (escape hatch when items raises)
• model - Server-reported model used
• usage - VLEmbeddingUsage with input_tokens / text_tokens / image_tokens (covers sampled video frames too) / image_count (excludes video frames) / duration (video seconds)
• warning - Optional server warning (string only; non-string payloads return None)

Response Methods:

• get_embeddings() - Get all vectors
• get_first_embedding() - Get first vector (useful in fusion mode)

🔧 Async Support

import asyncio
from octen import AsyncOcten, ChatMessage

async def main():
    async with AsyncOcten(api_key="your-api-key") as client:
        # Concurrent chat requests
        task1 = client.chat.create(
            model="openai/gpt-5.4",
            messages=[ChatMessage(role="user", content="Explain deep learning")],
        )
        task2 = client.chat.create(
            model="anthropic/claude-sonnet-4.6",
            messages=[ChatMessage(role="user", content="Explain reinforcement learning")],
        )
        r1, r2 = await asyncio.gather(task1, task2)
        print(r1.text)
        print(r2.text)

        # Async streaming
        stream = await client.chat.create(
            model="openai/gpt-5.4",
            messages=[ChatMessage(role="user", content="Hello!")],
            stream=True
        )
        async for event in stream:
            if event.type == "content" and event.choices:
                print(event.choices[0].delta.content or "", end="", flush=True)

        # Search, extract, and embeddings also work async
        results = await client.search.search(query="AI")
        extracted = await client.extract.extract(urls=["https://example.com"])
        embedding = await client.embedding.create(input=["Hello"], model="octen-embedding-4b")

asyncio.run(main())

⚠️ Error Handling

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

with Octen(api_key="your-api-key") as client:
    try:
        response = client.chat.create(
            model="openai/gpt-5.4",
            messages=[ChatMessage(role="user", content="Hello")]
        )
    except OctenAuthenticationError:
        print("Invalid or missing API key")
    except OctenRateLimitError as e:
        print(f"Rate limited — retry after {e.retry_after}s")
    except OctenStreamError as e:
        print(f"Stream error: {e.message} (code {e.code})")
    except OctenTimeoutError as e:
        print(f"Request timed out after {e.timeout}s")
    except OctenAPIError as e:
        print(f"API error {e.status_code}: {e.message}")

What each exception means

Every exception the SDK raises is one of its own. A bad value never surfaces as a pydantic.ValidationError, a transport failure never as an httpx exception, and a malformed response never as an internal TypeError. The original exception is kept on __cause__ when you need it.

Exception	Raised when
OctenValidationError	A parameter value is out of contract, caught before any request. .errors carries the field-level detail.
OctenAuthenticationError	HTTP 401.
OctenRateLimitError	HTTP 429. .retry_after is a float — seconds parsed from Retry-After, whether the server sent delta-seconds or an HTTP-date — or None when the header was absent or unusable.
OctenAPIError	Any other failed response, and a 2xx response that cannot be trusted: a body that is not a non-empty JSON object, an envelope whose numeric code is non-zero, or a body that cannot build the endpoint's response model. In those cases .status_code is 200.
OctenStreamError	A stream carried an error event, ended before it was complete, or contained a malformed event.
OctenTimeoutError	A deadline expired.
OctenConnectionError	The connection failed or the peer went away mid-response.

A wrong call shape — an unknown keyword, a duplicated or extra positional argument — raises Python's own TypeError, sync and async alike. That is a bug in the call, not invalid data.

Timeouts

Precedence is: a per-call timeout beats an explicit client timeout, which beats the endpoint default.

Octen(api_key=...)                        # 30s, and 60s for chat
Octen(api_key=..., timeout=10)            # 10s everywhere, chat included
client.chat.create(..., timeout=120)      # 120s for this call

extract(timeout=...) is not any of those: it is a field of the request body — the upstream per-URL fetch budget, an integer from 1 to 60 seconds. The local socket deadline for an extract call is request_timeout, and it should be comfortably larger than the upstream budget.

Streaming

A stream is complete only when the server says so: a [DONE] sentinel, a typed finish event, or an OpenAI-style finish_reason. A stream that simply stops raises OctenStreamError rather than looking like a shorter answer, and a malformed event raises rather than being skipped. An in-stream error event is yielded first and then raised, so content you have already rendered is still yours — but the iteration does fail, so do not treat "some events arrived" as success.

Requests carry exactly what you asked for

• A parameter left unset is omitted from the JSON body; nothing is ever sent as null.
• An explicit False is sent (include_images=False, enable_fusion=False).
• An explicit tools=[] is sent as [] — "no tools for this turn" is a different statement from not configuring tools, so the two produce different requests.
• embedding.create accepts a single string for convenience, but the wire format is always an array. Empty input — "", " ", [], or a list with a blank element — fails before the request.
• Date parameters are strings and are sent verbatim; the SDK does not reformat them.

🧪 Development

Install Development Dependencies

# Install development version from source
pip install -e ".[dev]"

Run Tests

pytest tests/

Code Formatting

black octen/
ruff check octen/ --fix

Type Checking

mypy octen/

📝 License

MIT License - See LICENSE file for details

🔗 Links

• Official Website
• API Documentation

📧 Support

For questions or help, please:

• Check the Documentation
• Email us at support@octen.ai