Steel

SYSTEM = """
You answer research-style questions by combining You.com APIs with a Steel cloud browser.
Prefer the cheap path first: youcom_search to find candidate URLs, then
youcom_contents to read them. Only call navigate, snapshot, or click_text when
the page is JS-rendered, login-walled, or you need to interact with filters,
toggles, or form fields.
"""
tools = [youcom_search, youcom_contents, navigate, snapshot, click_text]

import os
import time
from typing import Any
import httpx
from langchain_core.tools import tool
YOU_BASE = "https://ydc-index.io/v1"
YOUCOM_API_KEY = os.environ["YOUCOM_API_KEY"]
def _headers() -> dict[str, str]:
    return {"X-API-Key": YOUCOM_API_KEY, "Content-Type": "application/json"}
@tool
async def youcom_search(query: str, count: int = 5) -> dict[str, Any]:
    """Search the web for candidate URLs before fetching page content."""
    start = time.perf_counter()
    async with httpx.AsyncClient(timeout=30) as client:
        response = await client.post(
            f"{YOU_BASE}/search",
            json={"query": query, "count": count},
            headers=_headers(),
        )
    response.raise_for_status()
    data = response.json()
    print(f"youcom_search: {(time.perf_counter() - start) * 1000:.0f}ms")
    return data
@tool
async def youcom_contents(urls: list[str]) -> dict[str, Any]:
    """Fetch clean Markdown for up to 10 URLs before escalating to a browser."""
    start = time.perf_counter()
    async with httpx.AsyncClient(timeout=60) as client:
        response = await client.post(
            f"{YOU_BASE}/contents",
            json={"urls": urls, "formats": ["markdown"]},
            headers=_headers(),
        )
    response.raise_for_status()
    data = response.json()
    print(f"youcom_contents: {(time.perf_counter() - start) * 1000:.0f}ms")
    return data

import os
import time
from langchain_core.tools import tool
from playwright.async_api import Page, async_playwright
from steel import Steel
STEEL_API_KEY = os.environ["STEEL_API_KEY"]
steel = Steel(api_key=STEEL_API_KEY)
_session = None
_playwright = None
_browser = None
_page: Page | None = None
async def _ensure_session() -> Page:
    global _session, _playwright, _browser, _page
    if _page is not None:
        return _page
    start = time.perf_counter()
    _session = steel.sessions.create()
    _playwright = await async_playwright().start()
    _browser = await _playwright.chromium.connect_over_cdp(
        f"{_session.websocket_url}&apiKey={STEEL_API_KEY}"
    )
    _page = _browser.contexts[0].pages[0]
    print(f"open_session: {(time.perf_counter() - start) * 1000:.0f}ms")
    return _page
@tool
async def navigate(url: str) -> str:
    """Open a URL in Steel when static extraction is not enough."""
    page = await _ensure_session()
    start = time.perf_counter()
    await page.goto(url, wait_until="domcontentloaded")
    print(f"navigate: {(time.perf_counter() - start) * 1000:.0f}ms")
    return page.url
@tool
async def snapshot() -> str:
    """Read visible text from the live page after JavaScript runs."""
    page = await _ensure_session()
    text = await page.locator("body").inner_text(timeout=10_000)
    print(f"snapshot: {len(text)} chars")
    return text[:20_000]
@tool
async def click_text(text: str) -> str:
    """Click the first visible element whose text matches the input."""
    page = await _ensure_session()
    await page.get_by_text(text).first.click(timeout=10_000)
    await page.wait_for_load_state("domcontentloaded")
    return f"Clicked {text!r}"

async def close_browser() -> None:
    if _browser is not None:
        await _browser.close()
    if _playwright is not None:
        await _playwright.stop()
    if _session is not None:
        steel.sessions.release(_session.id)

import asyncio
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", SYSTEM),
        ("human", "{input}"),
        MessagesPlaceholder("agent_scratchpad"),
    ]
)
model = ChatOpenAI(model="gpt-5-mini")
agent = create_tool_calling_agent(model, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, max_iterations=10, verbose=False)
async def main() -> None:
    try:
        result = await executor.ainvoke(
            {"input": "Find the latest pricing page for a database company and summarize its free tier."}
        )
        print(result["output"])
    finally:
        await close_browser()
asyncio.run(main())