Quickstart (Python)

1
import asyncio
2
import os
3
from typing import Optional
4

5
from agents import Agent, Runner, function_tool
6
from dotenv import load_dotenv
7
from playwright.async_api import Browser, Page, async_playwright
8
from pydantic import BaseModel, Field
9
from steel import Steel
10

11
load_dotenv()
12

13
STEEL_API_KEY = os.getenv("STEEL_API_KEY") or "your-steel-api-key-here"
14
steel = Steel(steel_api_key=STEEL_API_KEY)
15

16
_session = None
17
_browser: Optional[Browser] = None
18
_page: Optional[Page] = None
19
_playwright = None
20

21

22
@function_tool
23
async def open_session() -> dict:
24
    """Open a Steel cloud browser session. Call exactly once, before anything else."""
25
    global _session, _browser, _page, _playwright
26
    _session = steel.sessions.create()
27
    _playwright = await async_playwright().start()
28
    _browser = await _playwright.chromium.connect_over_cdp(
29
        f"{_session.websocket_url}&apiKey={STEEL_API_KEY}"
30
    )
31
    ctx = _browser.contexts[0]
32
    _page = ctx.pages[0] if ctx.pages else await ctx.new_page()
33
    return {"session_id": _session.id, "live_view_url": _session.session_viewer_url}
34

35

36
@function_tool
37
async def navigate(url: str) -> dict:
38
    """Navigate the open session to a URL and wait for the page to load."""
39
    if _page is None:
40
        raise RuntimeError("open_session first.")
41
    await _page.goto(url, wait_until="domcontentloaded", timeout=45_000)
42
    return {"url": _page.url, "title": await _page.title()}
43

44

45
@function_tool
46
async def snapshot(max_chars: int = 4_000, max_links: int = 50) -> dict:
47
    """Return a readable snapshot of the current page: title, URL, visible
48
    text (capped), and a list of links. Call BEFORE extract so the agent
49
    never has to guess CSS selectors.
50
    """
51
    if _page is None:
52
        raise RuntimeError("open_session first.")
53
    return await _page.evaluate(
54
        """({maxChars, maxLinks}) => {
55
            const text = (document.body.innerText || '').slice(0, maxChars);
56
            const links = Array.from(document.querySelectorAll('a[href]'))
57
                .slice(0, maxLinks)
58
                .map((a) => ({
59
                    text: (a.innerText || a.textContent || '').trim().slice(0, 120),
60
                    href: a.href,
61
                }))
62
                .filter((l) => l.text && l.href);
63
            return { url: location.href, title: document.title, text, links };
64
        }""",
65
        {"maxChars": max_chars, "maxLinks": max_links},
66
    )
67

68

69
class FieldSpec(BaseModel):
70
    name: str
71
    selector: str = Field(
72
        description="CSS selector relative to the row. Empty string reads the row itself."
73
    )
74
    attr: Optional[str] = Field(
75
        default=None,
76
        description="Optional attribute to read instead of innerText (e.g. 'href').",
77
    )
78

79

80
@function_tool
81
async def extract(
82
    row_selector: str, fields: list[FieldSpec], limit: int = 10
83
) -> dict:
84
    """Extract structured rows from the current page using CSS selectors.
85
    Prefer calling snapshot() first to confirm the page structure.
86
    """
87
    if _page is None:
88
        raise RuntimeError("open_session first.")
89
    fields_json = [{"name": f.name, "selector": f.selector, "attr": f.attr} for f in fields]
90
    items = await _page.evaluate(
91
        """({rowSelector, fields, limit}) => {
92
            const rows = Array.from(
93
                document.querySelectorAll(rowSelector)
94
            ).slice(0, limit);
95
            return rows.map((row) => {
96
                const item = {};
97
                for (const f of fields) {
98
                    const el = f.selector ? row.querySelector(f.selector) : row;
99
                    if (!el) { item[f.name] = ''; continue; }
100
                    item[f.name] = f.attr
101
                        ? (el.getAttribute(f.attr) || '').trim()
102
                        : (el.innerText || el.textContent || '').trim();
103
                }
104
                return item;
105
            });
106
        }""",
107
        {"rowSelector": row_selector, "fields": fields_json, "limit": limit},
108
    )
109
    return {"count": len(items), "items": items}