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