octen 0.3.0

Official Python SDK for Octen API - Web Search, URL Extract, Text Embeddings, and LLM Chat

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, query-driven highlights, and media (images / videos / audio / favicon)
• 💬 Multi-model Chat — access 10+ LLMs (GPT, Claude, Gemini, Kimi, MiniMax) through a single unified API
• 🧮 Text Embeddings — convert text into high-quality vector representations
• ⚡ 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', '')}")

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

with Octen(api_key="your-api-key") as client:
    response = client.chat.create(
        model="openai/gpt-5.4",
        messages=[ChatMessage(role="user", content="Hello!")],
        web_search="on"
    )
    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)}")

🔍 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,
        search_type="semantic",  # Semantic search
        include_domains=["github.com", "arxiv.org"],  # Search only these domains
        start_time="2024-01-01T00:00:00Z",  # Time filtering
        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"Actual search type: {response.search_type}")
    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_favicon=True,
        include_images=True,
        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:
                print(f"    favicon: {item.favicon}")
        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.

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.

💬 Chat API

Non-streaming

from octen import Octen, ChatMessage, WebSearchOptions

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?"),
        ],
        web_search="on",
        web_search_options=WebSearchOptions(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

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,
        web_search="on"
    ):
        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"),
        web_search="off"
    )
    print(response.text)

Web Search with Full Page Content

from octen import Octen, ChatMessage, WebSearchOptions
from octen.models.chat import ChatFullContentOptions

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?")],
        web_search="on",
        web_search_options=WebSearchOptions(
            safesearch="off",
            full_content=ChatFullContentOptions(enable=True, max_tokens=1000)
        )
    )
    print(f"Full content tokens: {response.usage.full_content_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"])

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
• count (int, optional): Number of results to return, range 1-100, default 5
• search_type (str, optional): Search type, options:
  • "auto" - Automatically select (default)
  • "keyword" - Keyword search
  • "semantic" - Semantic search
• include_domains (List[str], optional): Include only results from these domains
• exclude_domains (List[str], optional): Exclude results from these domains
• include_text (List[str], optional): Results must contain these texts
• exclude_text (List[str], optional): Results must exclude these texts
• time_basis (str, optional): Time basis, options: "auto", "published", "crawled"
• 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 configuration
• format (str, optional): Content format, options: "text", "markdown"
• safesearch (str, optional): Safe search, options: "off", "strict" (default)
• full_content (FullContentOptions, optional): Full content options configuration
• timeout (float, optional): Request timeout in seconds

Returns: SearchResponse object

Response Properties:

• results - List of search results
• query - The actual query used
• search_type - The actual search type used
• 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. Server clamps into [300, 31_536_000] (5 min – 1 year) and defaults to 86_400 (24h) when omitted
• 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 (default False)
• include_favicon (bool, optional): Return favicon URL (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

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}
• total_urls (int | None) — Convenience accessor
• successful_urls (int | None) — Number of URLs with status="success" in this batch
• 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) — When include_favicon=True
• images / videos / audio (List[Any] | None) — Passthrough media objects; schema is upstream-defined
• 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
• web_search (str, optional): "on" to augment with live web search, "off" to disable
• web_search_options (WebSearchOptions, optional): Fine-grained search configuration
  • safesearch (str): "off" or "strict" (default "off")
  • count (int): Number of search results, range 1-100
  • country (str): Country code for localised results (e.g. "CN")
  • include_domains / exclude_domains (List[str]): Domain filtering
  • include_text / exclude_text (List[str]): Text filtering
  • time_basis (str): "auto", "published", or "crawled"
  • start_time / end_time (str): ISO 8601 time range
  • format (str): "text" or "markdown"
  • full_content (ChatFullContentOptions): Full page content options
  • highlight (ChatHighlightOptions): Highlight snippet options
• 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]
• 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: "low", "medium", or "high"
• 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 | dict], optional): Tool/function definitions available to the model
• tool_choice (str | dict, optional): "none", "auto", "required", or a dict specifying a particular tool
• 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
• text - Convenience accessor for the first choice's content
• usage - Usage object (prompt_tokens, completion_tokens, total_tokens, num_search_queries, reasoning_tokens)
• search_results - List of ChatSearchResult (when web_search="on")
• 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 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

🔧 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")],
            web_search="off"
        )
        task2 = client.chat.create(
            model="anthropic/claude-sonnet-4.6",
            messages=[ChatMessage(role="user", content="Explain reinforcement learning")],
            web_search="off"
        )
        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}")

🧪 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 [e-mail]