tinyfish 0.2.5


pip install tinyfish

  Latest version

Released: Apr 03, 2026

Project Links

Meta
Requires Python: >=3.11

Classifiers

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.

Method Description Returns Blocks?
agent.run() Run an automation, wait for the result AgentRunResponse Yes
agent.queue() Start an automation, return immediately AgentRunAsyncResponse No
agent.stream() Stream live SSE events as the agent works AgentStream No
runs.get() Retrieve a single run by ID Run
runs.list() List runs with filtering, sorting, pagination RunListResponse

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
    ),
)

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

Returns AgentRunResponse:

Field Type Description
status RunStatus COMPLETED, FAILED, etc.
run_id str | None Unique run identifier
result dict | None Extracted data (None if failed)
error RunError | None Error details (None if succeeded)
num_of_steps int Number of steps the agent took
started_at datetime | None When the run started
finished_at datetime | None When the run finished

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.

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:

Field Type Description
run_id str | None Run ID to poll with runs.get()
error RunError | None Error if queuing itself failed

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: STARTEDSTREAMING_URLPROGRESS (repeated) → COMPLETE.

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


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.streaming_url)          # live browser URL (while RUNNING)
print(run.browser_config)         # proxy/browser settings that were used

Returns Run:

Field Type Description
run_id str Unique identifier
status RunStatus PENDING, RUNNING, COMPLETED, FAILED, CANCELLED
goal str The goal that was given
result dict | None Extracted data (None if not completed)
error RunError | None Error details (None if succeeded)
streaming_url str | None Live browser URL (available while running)
browser_config BrowserConfig | None Proxy/browser settings used
created_at datetime When the run was created
started_at datetime | None When execution started
finished_at datetime | None When execution finished

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:

Field Type Description
data list[Run] List of runs
pagination.total int Total runs matching filters
pagination.has_more bool Whether more pages exist
pagination.next_cursor str | None Pass to cursor= for the next page

See the Pagination Guide for full pagination loop examples.


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 five methods (agent.run(), agent.queue(), agent.stream(), runs.get(), runs.list()) 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="https://agent.tinyfish.ai",  # optional — default shown
    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,
)

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

Wheel compatibility matrix

Platform Python 3
any

Files in release

Extras: None
Dependencies:
httpx (>=0.27.0)
pydantic (>=2.0.0)
tenacity (>=8.0.0)