tinyfish 0.2.5

Official Python SDK for the TinyFish API

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: STARTED → STREAMING_URL → PROGRESS (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

• 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