webiq 0.1.0

Web IQ API Python SDK

Web IQ Python SDK

Official Python SDK for Web IQ APIs.

Installation

pip install webiq

Quick Start

import os
from webiq import WebIQClient
from webiq.types import BrowseContentFormat, ContentFormat

with WebIQClient(api_key="your-api-key") as client:
    # Web search with content format
    response = client.web.search(
        "Python programming",
        max_results=5,
        content_format=ContentFormat.html,
    )
    for result in response.webResults or []:
        print(f"{result.title}: {result.url}")

    # News search
    news = client.news.search("technology", max_results=5)
    for item in news.newsResults or []:
        print(f"{item.title} - {item.source}")

    # Video search
    videos = client.videos.search("machine learning tutorial", max_results=5)
    for video in videos.videoResults or []:
        print(f"{video.title} ({video.length})")

    # Browse a URL in markdown format
    page = client.browse.fetch("https://www.microsoft.com", content_format=BrowseContentFormat.markdown)
    print(page.content)

    # Classic search with multiple answer types
    classic = client.classic.search(
        query="Artificial intelligence trends",
        response_filter=["webResults"],
    )
    # response_filter selects which answer types come back; the matching
    # attributes (e.g. classic.webResults) are populated dynamically.
    print(f"Web Results ({len(classic.webResults or [])}):")
    for r in classic.webResults or []:
        print(f"  - {r['title']}: {r['url']}")

Authentication

API Key

import os
from webiq import WebIQClient

client = WebIQClient(api_key=os.environ["WEBIQ_API_KEY"])

EntraID (Azure AD)

Pass any azure-identity TokenCredential:

from azure.identity import DefaultAzureCredential
from webiq import WebIQClient

client = WebIQClient(credential=DefaultAzureCredential())

Async Support

import asyncio
from webiq import WebIQAsyncClient

async def main():
    async with WebIQAsyncClient(api_key="your-api-key") as client:
        web, news = await asyncio.gather(
            client.web.search("async Python"),
            client.news.search("programming"),
        )
        for result in web.webResults or []:
            print(result.title)

asyncio.run(main())

API Reference

Web Search

response = client.web.search(
    query="search query",         # Required (1-1000 chars)
    max_results=10,               # 1-50, default 10
    language="en",                # ISO 639-1 code
    region="US",                  # Country/region code
    location="lat:40.7;long:-74.0",  # Optional
    content_format=ContentFormat.html,
    max_length=10000,             # Max content chars (1-500000)
)
# response.webResults → list of {title, url, content, lastUpdatedAt, ...}

News Search

response = client.news.search(
    query="search query",         # Required (1-1000 chars)
    max_results=10,               # 1-20, default 10
    language="en",
    region="US",
    location="lat:40.7;long:-74.0",  # Optional
    content_format=ContentFormat.text,
    max_length=10000,
)
# response.newsResults → list of {title, url, content, snippet, source, ...}

Video Search

response = client.videos.search(
    query="search query",         # Required (1-1000 chars)
    max_results=30,               # 1-30, default 30
    language="en",
    region="US",
    enable_playlist=True,
    freshness="month",            # week, month, year
)
# response.videoResults → list of {title, url, length, viewCount, moments, ...}
# response.playlists → list of {title, videos, ...}

Browse

response = client.browse.fetch(
    url="https://www.microsoft.com",    # Required
    max_length=10000,                   # Max content chars (1-500000)
    live_crawl="fallback",              # "none" (default) | "fallback" | "force"
    include_web_links=True,
    include_image_links=True,
    render_dynamic_pages=False,
    content_format=BrowseContentFormat.markdown,
)
# response → {url, title, content, isAdult, retryAfter, traceId, ...}

Image Search

from webiq.types import ImageAspectRatio, ImageSize, SafeSearch

response = client.images.search(
    query="search query",         # Required (1-1000 chars)
    max_results=30,               # 1-30, default 30
    language="en",
    region="US",
    aspect_ratio=ImageAspectRatio.wide,  # square, wide, tall
    image_size=ImageSize.large,          # small, medium, large, extraLarge
    safe_search=SafeSearch.strict,       # off, strict
    watermark_free=True,
)
# response.imageResults → list of {title, url, hostPageUrl, caption, width, height, thumbnailUrl, ...}

Classic Search

Given a query, classic search will search and retrieve content from webpages, images, videos, news, weather, sports, etc.

from webiq.types import ContentFormat, SafeSearchMode

response = client.classic.search(
    query="search query",         # Required (1-1000 chars)
    max_answer_types=6,           # 1-6, default 6
    language="en",                # ISO 639-1/2 or BCP 47
    region="US",                  # Country/region code
    location="lat:40.7;long:-74.0",  # Optional
    max_results_web=10,           # 1-50, default 10
    max_length=10000,             # Max content chars (1-500000)
    content_format=ContentFormat.html,
    freshness="month",            # day, week, month, year, or date range
    response_filter=["webResults", "newsResults"],  # Answer types to include
    safe_search=SafeSearchMode.moderate,   # off, moderate, strict
)
# response.querySignals → {originalQuery, normalizedQuery, isDefensive, isAdult, isNav, isFresh}
# response.traceId → str
# Additional dynamic fields for different answer types are exposed by
# Pydantic's ``extra="allow"`` and come back as raw ``dict`` values — use
# index access (``r['title']``), not attribute access (``r.title``).
# response filter: Allowed types: 'microAnswerResults', 'videoResults', 'dictionaryResults', 'packageTrackingResults', 'lyricsResults', 'movieResults', 'eventResults', 'jobResults', 'weatherResults', 'appResults', 'techHelpResults', 'directionResults', 'factCarouselResults', 'factResults', 'healthResults', 'recipeResults', 'questionAndAnswerResults', 'computationResults', 'prayerTimeResults', 'financeResults', 'mapResults', 'webResults', 'newsResults', 'imageResults', 'sportsResults', 'entityResults', 'realEstateResults', 'timeZoneResults', 'travelResults', 'placeResults'

Enum Types

Some parameters require enum values instead of plain strings. Import them from webiq.types:

from webiq.types import BrowseContentFormat, ContentFormat

# ContentFormat: format of returned content for web, news, and classic search
ContentFormat.passage    # Selected passages only (plain text)
ContentFormat.text       # Full page text (plain text)
ContentFormat.html       # HTML format (default for web search)
ContentFormat.markdown   # Markdown format

# BrowseContentFormat: format for browse (no passage option)
BrowseContentFormat.text       # Full page text
BrowseContentFormat.html       # HTML format (default)
BrowseContentFormat.markdown   # Markdown format

# Usage in web search
response = client.web.search("query", content_format=ContentFormat.markdown)

# Usage in browse
page = client.browse.fetch("https://www.microsoft.com", content_format=BrowseContentFormat.html)

Configuration

Timeout and Retry

The client accepts timeout (seconds) and retry (a RetryPolicy) as top-level keyword arguments. Per-call timeout on each resource method overrides the client-level default.

from webiq import WebIQClient, RetryPolicy

client = WebIQClient(
    api_key="your-api-key",
    timeout=10.0,                  # seconds (default: 10.0)
    retry=RetryPolicy(
        max_retries=2,             # defaults shown
        base_delay_s=0.25,
        max_delay_s=4.0,
    ),
)

# Per-call overrides
response = client.web.search("query", language="de", region="DE", timeout=30.0)

Customized HTTP client (proxy, TLS, ...)

For advanced HTTP settings — proxy, custom TLS, connection pooling, etc. — pass a pre-configured httpx.Client via http_client. Note: the SDK does not close caller-owned clients.

import httpx
from webiq import WebIQClient, WebIQAsyncClient

# sync:
http_client = httpx.Client(
    base_url="https://api.microsoft.ai/v3",
    proxy="http://proxy:8080",
)
client = WebIQClient(api_key="your-api-key", http_client=http_client)
http_client = httpx.AsyncClient(
    base_url="https://api.microsoft.ai/v3",
    proxy="http://proxy:8080",
)
client = WebIQAsyncClient(api_key="your-api-key", http_client=http_client)

Telemetry

from webiq import WebIQClient, TelemetryEvent

def on_request(event: TelemetryEvent):
    print(f"{event.method} {event.path} → {event.status_code} ({event.elapsed_ms}ms)")

client = WebIQClient(api_key="your-api-key", telemetry_hook=on_request)

Error Handling

The SDK raises specific exception types for different error conditions. All exceptions inherit from WebIQError.

Exception Hierarchy

Exception	HTTP Status	When
AuthenticationError	401	Invalid or missing API key
PermissionDeniedError	403	Authenticated but not authorized for the resource
RateLimitError	429, 430	Rate limit / concurrent-request limit exceeded
APIStatusError	400, 404, 500, 503, 504, ...	All other HTTP errors
APIConnectionError	—	Network issues, DNS failures, timeouts
WebIQError	—	Base class for all SDK errors

PermissionDeniedError and RateLimitError are both subclasses of APIStatusError, so a broader except APIStatusError clause still catches them.

Rate limits are never auto-retried

The SDK does not automatically retry 429 (rate limit) or 430 (concurrent-request limit) responses. As soon as the API returns one, the transport raises RateLimitError so your application can decide what to do — back off, queue the request, surface it to the user, etc. Generic retry settings (RetryPolicy.retry_on_status, max_retries) do not apply to rate limits.

The server reports the back-off hint in the response body as the retryAfter field — typically a duration with an s suffix (e.g. "30s", "60s"). The SDK surfaces that value unchanged on error.retry_after.

import time
from webiq import WebIQClient, RateLimitError

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

try:
    response = client.web.search("test")
except RateLimitError as e:
    # e.retry_after is the server-provided value from the response body
    # (e.g. "60s"). The SDK does not parse or normalize it for you.
    print(f"Rate limited. Retry after: {e.retry_after}")
    # Your retry strategy lives here — the SDK will not retry for you.

Basic Error Handling

from webiq import (
    WebIQClient,
    WebIQError,
    APIConnectionError,
    APIStatusError,
    AuthenticationError,
    PermissionDeniedError,
    RateLimitError,
)

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

try:
    response = client.web.search("test")
except PermissionDeniedError as e:
    # 403 — authenticated, but not allowed to call this resource
    print(f"Forbidden (HTTP {e.status_code}): {e}")
except AuthenticationError as e:
    # 401 — invalid or missing API key
    print(f"Auth failed (HTTP {e.status_code}): {e}")
except RateLimitError as e:
    # 429 or 430 — rate limit / concurrent-request limit (never auto-retried)
    print(f"Rate limited. Retry after: {e.retry_after}")
except APIStatusError as e:
    # Other HTTP errors (400, 404, 500, 503, 504, etc.)
    print(f"API error (HTTP {e.status_code}): {e}")
except APIConnectionError as e:
    # Network issues
    print(f"Connection failed: {e}")
except WebIQError as e:
    # Catch-all for any SDK error
    print(f"SDK error: {e}")

Inspecting Error Details

All APIStatusError exceptions (including PermissionDeniedError and RateLimitError) expose the full error response body. Most input-validation problems (e.g. max_results out of range, empty query) are caught client-side by Pydantic and raise pydantic.ValidationError before any request is sent; the pattern below applies when the server rejects a request (e.g. a 4xx the SDK couldn't pre-validate, or a transient 5xx). browse.fetch is a convenient way to trigger one — a missing or filtered URL surfaces as a structured error:

from webiq import WebIQClient
from webiq.errors import APIStatusError

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

try:
    response = client.browse.fetch("https://www.not-microsoft.com")
except APIStatusError as e:
    print(f"Status: {e.status_code}")                          # e.g. 404
    print(f"Message: {e}")                                      # e.g. "No result is found"

    # Full error body from the API response
    if isinstance(e.body, dict):
        print(f"Error code: {e.body.get('errorCode')}")        # e.g. "BrowseApiDocNotFound"
        print(f"Category: {e.body.get('errorCategory')}")      # e.g. "UserError"
        print(f"Details: {e.body.get('technicalDetails')}")    # e.g. "NotFound"
        print(f"Trace ID: {e.body.get('traceId')}")            # for debugging with support
        print(f"Retry after: {e.body.get('retryAfter')}")      # for retryable errors

License

MIT