TinyFish Python SDK

The official Python SDK for TinyFish

Installation

pip install tinyfish

Requires Python 3.11+.

Get your API key

Sign up and grab your key at agent.tinyfish.ai/api-keys.

Quickstart

from tinyfish import TinyFish

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

response = client.agent.run(
    goal="What is the current Bitcoin price?",
    url="https://www.coinbase.com/price/bitcoin",
)
print(response.result)

Or set the TINYFISH_API_KEY environment variable and omit api_key:

client = TinyFish()

Methods

Every method below is available on both TinyFish (sync) and AsyncTinyFish (async). Async versions have the same signatures — just await them.

agent.run() — block until done

Sends the automation and waits for it to finish. Returns the full result in one shot.

from tinyfish import TinyFish, RunStatus, BrowserProfile, ProxyConfig, ProxyCountryCode

client = TinyFish()

response = client.agent.run(
    goal="Extract the top 5 headlines",              # required — what to do on the page
    url="https://news.ycombinator.com",              # required — URL to open
    browser_profile=BrowserProfile.STEALTH,          # optional — "lite" (default) or "stealth"
    proxy_config=ProxyConfig(                        # optional — proxy settings
        enabled=True,
        country_code=ProxyCountryCode.US,            # optional — US, GB, CA, DE, FR, JP, AU
    ),
    output_schema={                                  # require structured output
        "type": "object",
        "properties": {
            "headline_count": {"type": "integer"},
            "top_headline": {"type": "string"},
        },
        "required": ["headline_count", "top_headline"],
    },
)

if response.status == RunStatus.COMPLETED:
    print(response.result)
else:
    print(f"Failed: {response.error.message}")

output_schema must be a top-level object. The SDK sends the schema to the API as-is; invalid schemas are rejected by the API before execution.

The same output_schema= keyword is available on client.agent.queue() and client.agent.stream(). Persisted runs return the requested contract back as run.output_schema.

Returns AgentRunResponse:

agent.queue() — fire and forget

Starts the automation in the background and returns a run_id immediately. Poll with runs.get() when you're ready for the result.

queue() accepts the same structured-output parameters as run(), including output_schema=....

import time
from tinyfish import TinyFish, RunStatus

client = TinyFish()

queued = client.agent.queue(
    goal="Extract the top 5 headlines",              # required — what to do on the page
    url="https://news.ycombinator.com",              # required — URL to open
    browser_profile=None,                            # optional — "lite" (default) or "stealth"
    proxy_config=None,                               # optional — proxy settings
)
print(f"Run started: {queued.run_id}")

# Poll for completion
while True:
    run = client.runs.get(queued.run_id)
    if run.status in (RunStatus.COMPLETED, RunStatus.FAILED):
        break
    time.sleep(5)

print(run.result)

Returns AgentRunAsyncResponse:

agent.stream() — real-time events

Opens a Server-Sent Events stream. You get live progress updates as the agent works, plus a WebSocket URL for a live browser preview.

from tinyfish import TinyFish, CompleteEvent, ProgressEvent

client = TinyFish()

with client.agent.stream(
    goal="Extract the top 5 headlines",              # required — what to do on the page
    url="https://news.ycombinator.com",              # required — URL to open
    browser_profile=None,                            # optional — "lite" (default) or "stealth"
    proxy_config=None,                               # optional — proxy settings
    on_started=lambda e: print(f"Started: {e.run_id}"),          # optional — called when run starts
    on_streaming_url=lambda e: print(f"Watch: {e.streaming_url}"),  # optional — called with live browser URL
    on_progress=lambda e: print(f"  > {e.purpose}"),             # optional — called on each step
    on_heartbeat=lambda e: None,                                 # optional — called on keepalive pings
    on_complete=lambda e: print(f"Done: {e.status}"),            # optional — called when run finishes
) as stream:
    for event in stream:
        # Callbacks fire automatically during iteration.
        # You can also inspect events directly:
        if isinstance(event, CompleteEvent):
            print(event.result_json)

Returns AgentStream — a context manager you iterate over. Events arrive in order: STARTED → STREAMING_URL → PROGRESS (repeated) → COMPLETE.

See the Streaming Guide for the full event lifecycle, event types, and advanced patterns.

fetch.get_contents() — clean content

Fetch extracted content from one or more URLs without a browser-agent run.

from tinyfish import TinyFish

client = TinyFish()

response = client.fetch.get_contents(
    [
        "https://example.com",
        "https://example.org",
    ],
    format="markdown",
    links=True,
    image_links=False,
    per_url_timeout_ms=45_000,
)

print(response.results)
print(response.errors)

fetch.get_contents() accepts 1 to 10 URLs. Set per_url_timeout_ms to apply an independent timeout budget to each URL in the batch; slow URLs return in errors with timeout while siblings can still complete.

Pass highlights={"query": "pricing tiers"} (beta, markdown format only) to get query-ranked HighlightSnippet lists on each result instead of the full text; set include_full_page_text=True to keep both. Accounts not enrolled in the beta get PermissionDeniedError.

Returns FetchResponse:

runs.get() — retrieve a single run

Fetch the full details of a run by its ID.

run = client.runs.get(
    "run_abc123",   # required — the run ID
)

print(run.status)   # PENDING, RUNNING, COMPLETED, FAILED, CANCELLED
print(run.result)
print(run.goal)
print(run.output_schema)         # requested structured-output contract, if one was provided
print(run.streaming_url)          # live browser URL (while RUNNING)
print(run.browser_config)         # proxy/browser settings that were used

Returns Run:

Raises: ValueError if run_id is empty. NotFoundError if no run exists with that ID.

runs.list() — list and filter runs

List runs with optional filtering, sorting, and cursor-based pagination. All parameters are optional.

from tinyfish import RunStatus, SortDirection

response = client.runs.list(
    status=RunStatus.COMPLETED,                      # optional — filter by status
    goal="headlines",                                # optional — filter by goal text
    created_after="2025-01-01T00:00:00Z",            # optional — ISO 8601 lower bound
    created_before="2025-12-31T23:59:59Z",           # optional — ISO 8601 upper bound
    sort_direction=SortDirection.DESC,                # optional — "asc" or "desc"
    limit=10,                                        # optional — max runs per page
    cursor=None,                                     # optional — pagination cursor from previous response
)

for run in response.data:
    print(f"{run.run_id} | {run.goal}")

# Pagination
if response.pagination.has_more:
    next_page = client.runs.list(cursor=response.pagination.next_cursor)

Returns RunListResponse:

See the Pagination Guide for full pagination loop examples.

search.query() — search the web

Returns ranked web search results with titles, snippets, and URLs.

from tinyfish import TinyFish

client = TinyFish()

response = client.search.query("FIFA")

print(response.query)
print(response.total_results)
print(response.results[0].title if response.results else "No results")

Optional parameters:

• location — country code for geo-targeted results (e.g. "US", "GB")
• language — language code (e.g. "en", "fr")
• page — page number, 0-indexed, max 10
• recency_minutes — freshness window in minutes (1 to 5256000)
• after_date / before_date — calendar date range in YYYY-MM-DD
• domain_type — result category: "web" (default), "news", or "research_paper"
• pub_year_min / pub_year_max — publication-year range, inclusive (0 to 9999). Only supported for domain_type="research_paper"

# geo-targeted
response = client.search.query("FIFA", location="US", language="en")

# freshness window
response = client.search.query("FIFA", recency_minutes=60)

# calendar date range
response = client.search.query("FIFA", after_date="2026-06-01", before_date="2026-06-18")

# domain type
response = client.search.query("FIFA", domain_type="news")
response = client.search.query("machine learning", domain_type="research_paper")

# publication-year range (research_paper only)
response = client.search.query(
    "transformer architecture", domain_type="research_paper", pub_year_min=2019, pub_year_max=2022
)

Filter validation rules:

• recency_minutes must be an integer from 1 to 5256000
• after_date and before_date must use YYYY-MM-DD
• recency_minutes cannot be combined with after_date or before_date
• if both dates are present, after_date must be less than or equal to before_date
• domain_type must be one of "web", "news", or "research_paper"
• pub_year_min and pub_year_max must be integers from 0 to 9999
• if both are present, pub_year_min must be less than or equal to pub_year_max

wallet.get() — get your balance

Returns the caller's current balance, auto-reload configuration, per-product rates, and any in-flight top-up.

from tinyfish import TinyFish, WalletAutoReloadState

client = TinyFish()

wallet = client.wallet.get()

print(f"{wallet.available_balance} {wallet.currency}")

if wallet.auto_reload and wallet.auto_reload.state == WalletAutoReloadState.ON:
    print(f"Auto-reload at {wallet.auto_reload.threshold}, tops up to {wallet.auto_reload.recharge_to}")

if wallet.pending_top_up:
    print(f"Pending top-up: {wallet.pending_top_up.amount}")

Returns WalletResponse:

auto_reload.state is one of WalletAutoReloadState.UNCONFIGURED, ON, OFF, PAUSED_PAYMENT_FAILED, or NEEDS_PAYMENT_METHOD. threshold and recharge_to are only present for the four configured states.

Accounts are on either wallet billing or a legacy subscription/credits plan. wallet.get() raises NotFoundError (404) with .code of "FEATURE_NOT_AVAILABLE" when the account is on the legacy plan. A wallet account that hasn't added money yet is not an error — it returns normally with an available_balance of "0".

from tinyfish import NotFoundError

try:
    wallet = client.wallet.get()
except NotFoundError as e:
    if e.code == "FEATURE_NOT_AVAILABLE":
        print("Account is on the legacy subscription/credits plan, not a wallet.")
    else:
        raise

Sync vs Async

Use AsyncTinyFish when you're in an async context (FastAPI, aiohttp, etc.):

Sync:

from tinyfish import TinyFish

client = TinyFish()
response = client.agent.run(goal="...", url="...")

Async:

from tinyfish import AsyncTinyFish

client = AsyncTinyFish()
response = await client.agent.run(goal="...", url="...")

All eight methods (agent.run(), agent.queue(), agent.stream(), fetch.get_contents(), runs.get(), runs.list(), search.query(), wallet.get()) work the same way — same parameters, just await-ed.

Configuration

Client options

client = TinyFish(
    api_key="your-api-key",         # optional — or set TINYFISH_API_KEY env var
    base_url=None,                  # optional — omit it: Search and Fetch then use
                                    # https://api.search.tinyfish.ai and
                                    # https://api.fetch.tinyfish.ai; any explicit value
                                    # keeps every product on it. Do not pass a product host
                                    # here — it serves at the root, so /v1/search would 404
    timeout=600.0,                  # optional — seconds (default: 600)
    max_retries=2,                  # optional — retry attempts (default: 2)
)

The SDK retries 408, 429, and 5xx errors automatically with exponential backoff (0.5s multiplier, max 8s wait).

Browser profiles

Control the browser environment with browser_profile:

• lite (default) — fast, lightweight. Good for most sites.
• stealth — anti-detection mode. Use for sites with bot protection.

from tinyfish import BrowserProfile

response = client.agent.run(
    goal="...",
    url="...",
    browser_profile=BrowserProfile.STEALTH,
)

Agent controls and capture

agent.run(), agent.queue(), and agent.stream() also accept agent_config, capture_config, webhook_url, use_profile, and profile_id:

from tinyfish import AgentConfig, CaptureConfig

response = client.agent.run(
    goal="...",
    url="...",
    agent_config=AgentConfig(mode="strict", cursor_style="standard", max_duration_seconds=300),
    capture_config=CaptureConfig(screenshots=True, html=True),
    webhook_url="https://example.com/tinyfish-webhook",
    use_profile=True,
    profile_id="prof_abc123def4567890",
)

• agent_config (AgentConfig) — mode, cursor_style, max_steps, max_duration_seconds
• capture_config (CaptureConfig) — elements, snapshots, screenshots, recording, html
• webhook_url — HTTPS URL notified on run lifecycle events
• profile_id — a specific Browser Context Profile; requires use_profile=True

elements and recording are gated capabilities: an account without them enabled gets a 403 saying the capability is not enabled.

Browser Context Profiles

Use Browser Context Profiles when a run should start from saved logged-in state. Pass use_profile=True for your default profile, or add profile_id for a specific profile. Pair with use_vault=True when TinyFish should repair stale sessions with saved credentials.

response = client.agent.run(
    goal="Summarize the dashboard",
    url="https://app.example.com/dashboard",
    use_profile=True,
    profile_id="prof_abc123def4567890",
    use_vault=True,
)

Proxy configuration

Route requests through a proxy, optionally pinned to a country:

from tinyfish import ProxyConfig, ProxyCountryCode

response = client.agent.run(
    goal="...",
    url="...",
    proxy_config=ProxyConfig(enabled=True, country_code=ProxyCountryCode.US),
)

Available countries: US, GB, CA, DE, FR, JP, AU.

See the Proxy & Browser Profiles Guide for more details.

Error handling

from tinyfish import TinyFish, AuthenticationError, RateLimitError, SDKError

client = TinyFish()

try:
    response = client.agent.run(goal="...", url="...")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError:
    print("Rate limited (retries exhausted)")
except SDKError:
    print("Something else went wrong")

The SDK automatically retries transient errors (408, 429, 5xx) up to max_retries times with exponential backoff. Non-retryable errors (401, 400, 404) raise immediately.

For the full exception hierarchy and internal architecture, see docs/internal/exceptions-and-errors-guide.md.

Guides

• Streaming Guide — event lifecycle, callbacks vs iteration, event type reference
• Proxy & Browser Profiles — stealth mode, proxy countries
• Pagination Guide — filtering, sorting, cursor-based pagination
• Exceptions & Error Handling (internal) — layer-by-layer architecture
• Testing Guide — running and writing tests