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
fetch.get_contents()	Fetch extracted page content without running a browser agent	FetchResponse	Yes
runs.get()	Retrieve a single run by ID	Run	—
runs.list()	List runs with filtering, sorting, pagination	RunListResponse	—
search.query()	Search the web for ranked results	SearchQueryResponse	Yes
monitors.create()	Create a scheduled search or fetch Monitor and run its baseline	MonitorCreateResponse	Yes
monitors.list() / get() / update() / pause() / resume() / run() / cancel()	Manage Monitors	Monitor and friends	—
research.run()	Run deep research, wait for the report	ResearchRunDetail	Yes
research.queue()	Queue a deep research run, return its ID	ResearchRunAsyncResponse	No
research.stream()	Start deep research, stream its events	ResearchStream	No
research.events()	Attach or reconnect to a run's events	ResearchStream	No
research.get() / list() / cancel()	Read, list, and cancel research runs	ResearchRunDetail and friends	—
wallet.get()	Get the caller's wallet balance, auto-reload, and rates	WalletResponse	—
profiles.setup.start()	Open a browser to sign in to a profile	ProfileSetupSession	—
profiles.setup.save()	Save the signed-in state into the profile	ProfileSaveResult	Yes
profiles.setup.cancel()	Close the setup browser without saving	ProfileSetupCancelled	—

---

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 — any ISO 3166-1 alpha-2 code, e.g. "IN"
    ),
    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:

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.

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:

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.

---

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:

Field	Type	Description
results	list[FetchResult]	Successfully fetched URLs
errors	list[FetchError]	URLs that failed to fetch or extract

---

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:

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)
output_schema	dict | None	JSON Schema contract originally requested for the run
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.

---

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

---

monitors — re-run a search or fetch on a schedule

A Monitor repeats a search query or page fetch on a cron schedule and reports what changed. create() and run() are billed.

from tinyfish import TinyFish

client = TinyFish()

created = client.monitors.create(
    "fetch",
    {"url": "https://example.com/pricing", "format": "markdown"},
    "0 9 * * *",
    purpose="A plan price changes",
)
monitor_id = created.monitor.id

client.monitors.pause(monitor_id)
client.monitors.update(monitor_id, schedule_cron="0 8 * * 1-5", webhook_url=None)  # None clears a field
client.monitors.resume(monitor_id)
run = client.monitors.run(monitor_id).run  # SearchMonitorRun or FetchMonitorRun
client.monitors.cancel(monitor_id)  # deletes the Monitor and its schedule

config is a SearchMonitorConfig / FetchMonitorConfig or a dict, and must match type. A 402 raises PaymentRequiredError; its wallet_url points to where to add money.

---

research — deep research

Same shape as agent, with keyword-only arguments: run() blocks, queue() returns at once, stream() yields events. Options: output_language, domain_type, after_date / before_date, recency_minutes, weak_sources_enabled, domain_filter, prior_run_id or session_id. mode is "deep" (default, 5–20 min) or "max" (5–45 min). Starts are billed and never retried, since a retry could bill a second run.

run = client.research.run(query="Compare long-duration energy storage approaches", mode="deep")
if run.status == "COMPLETED" and run.deep_result:  # partial reports survive failures
    print(run.deep_result.result, run.deep_result.citations)

with client.research.stream(
    query="Compare long-duration energy storage approaches",
    on_progress=lambda e: print(e.event),  # also on_created, on_heartbeat, on_final_result, on_error
) as stream:
    for event in stream:
        if isinstance(event, ResearchFinalResultEvent):
            print(event.result)

run_id = client.research.queue(query="Compare long-duration energy storage approaches", mode="deep").research_run_id
run = client.research.get(run_id)  # ResearchRunDetail: status, deep_result, research_plan, evidence
client.research.list(status="COMPLETED", limit=20)
client.research.cancel(run_id)  # cancelled=False if it already finished

events(run_id) attaches or reconnects to a run and yields typed events until done. It has no overall time limit, resumes a dropped connection from the last event (Last-Event-ID), and leaving the with block detaches without cancelling. Streams have a 60s idle timeout and yield each event once; unknown event names are skipped, as in agent.stream(). Interrupting run() (Ctrl+C or task cancellation) only stops reading, as in agent; call cancel(run_id) to stop the run.

with client.research.events(run_id) as stream:  # `async with` / `async for` on AsyncTinyFish
    for event in stream:
        print(event.event)  # created, plan_updated, ..., final_result, done

---

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:

Field	Type	Description
available_balance	str	Current spendable balance. Negative means the wallet is overdrawn.
currency	str	ISO 4217 currency code, e.g. USD
as_of	str	ISO timestamp the wallet was read
auto_reload	WalletAutoReload | None	None means the auto-reload read failed — not the same as state == "unconfigured", which means the read succeeded and nothing is set
pending_top_up	WalletPendingTopUp | None	None means no in-flight top-up, or the ledger read failed
rates	WalletRates | None	None means the rates read failed, or the wallet has no rate contract yet

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

---

profiles.setup — sign in once, reuse the session

Start a setup browser for a Browser Context Profile, sign in through cdp_url, then save. Runs that pass profile_id start from that signed-in state.

from playwright.sync_api import sync_playwright
from tinyfish import TinyFish

client = TinyFish()

session = client.profiles.setup.start("prof_abc123def4567890", url="https://github.com/login")
with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(session.cdp_url)
    page = browser.contexts[0].pages[0]
    page.fill("#login_field", "me@example.com")
    page.fill("#password", "...")
    page.click("input[type=submit]")
    page.wait_for_url("https://github.com/")

result = client.profiles.setup.save(
    "prof_abc123def4567890",
    session_id=session.session_id,
    signed_in_sites=["github.com"],
)
print(result.cookie_count, result.sign_in)

cancel(profile_id, session_id=...) closes the browser without saving. Sessions expire at session.expires_at; a save after that returns 409.

---

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