tinyfish 0.3.0

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

---

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:

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.

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"

# 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")

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"

---

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 seven methods (agent.run(), agent.queue(), agent.stream(), fetch.get_contents(), runs.get(), runs.list(), search.query()) 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,
)

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