Quickstart

This guide walks you through connecting Agno with Steel by adding a Playwright-powered Steel toolkit and running an agent that browses and extracts content from live websites.

Prerequisites

Make sure you have:

  • Python 3.11+

  • Steel API key (get one at app.steel.dev)

  • (Optional) OpenAI API key if your Agno setup uses OpenAI models

Step 1: Project setup

Create and activate a virtual environment, then install dependencies:

Terminal
# Create project
mkdir steel-agno-starter
cd steel-agno-starter
# (Recommended) Create & activate a virtual environment
python3 -m venv .venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate
# Create files
touch main.py .env
# Install dependencies
pip install agno steel-sdk python-dotenv playwright

Create a .env file with your keys and a default task:

ENV
.env
1
STEEL_API_KEY=your_steel_api_key_here
2
OPENAI_API_KEY=your_openai_api_key_here # optional, if your Agno model needs it
3
TASK=Go to https://quotes.toscrape.com and: 1. Get the first 3 quotes with authors 2. Navigate to page 2 3. Get 2 more quotes from page 2

Step 2: Add a Steel toolkit and run an Agno Agent

First, define a toolkit that wraps Steel’s browser sessions and Playwright.

Python
main.py
1
import os
2
import json
3
from typing import Any, Dict, List, Optional
4
from agno.tools import Toolkit
5
from agno.utils.log import log_debug, logger
6
from playwright.sync_api import sync_playwright
7
from steel import Steel
8
9
10
class SteelTools(Toolkit):
11
def __init__(
12
self,
13
api_key: Optional[str] = None,
14
**kwargs,
15
):
16
"""Initialize SteelTools.
17
18
Args:
19
api_key (str, optional): Steel API key (defaults to STEEL_API_KEY env var).
20
"""
21
self.api_key = api_key or os.getenv("STEEL_API_KEY")
22
if not self.api_key:
23
raise ValueError(
24
"STEEL_API_KEY is required. Please set the STEEL_API_KEY environment variable."
25
)
26
27
self.client = Steel(steel_api_key=self.api_key)
28
29
self._playwright = None
30
self._browser = None
31
self._page = None
32
self._session = None
33
self._connect_url = None
34
35
tools: List[Any] = []
36
tools.append(self.navigate_to)
37
tools.append(self.screenshot)
38
tools.append(self.get_page_content)
39
tools.append(self.close_session)
40
41
super().__init__(name="steel_tools", tools=tools, **kwargs)
42
43
def _ensure_session(self):
44
"""Ensures a Steel session exists, creating one if needed."""
45
if not self._session:
46
try:
47
self._session = self.client.sessions.create() # type: ignore
48
if self._session:
49
self._connect_url = f"{self._session.websocket_url}&apiKey={self.api_key}" # type: ignore
50
log_debug(f"Created new Steel session with ID: {self._session.id}")
51
except Exception as e:
52
logger.error(f"Failed to create Steel session: {str(e)}")
53
raise
54
55
def _initialize_browser(self, connect_url: Optional[str] = None):
56
"""
57
Initialize browser connection if not already initialized.
58
Use provided connect_url or ensure we have a session with a connect_url
59
"""
60
if connect_url:
61
self._connect_url = connect_url if connect_url else "" # type: ignore
62
elif not self._connect_url:
63
self._ensure_session()
64
65
if not self._playwright:
66
self._playwright = sync_playwright().start() # type: ignore
67
if self._playwright:
68
self._browser = self._playwright.chromium.connect_over_cdp(self._connect_url)
69
context = self._browser.contexts[0] if self._browser else ""
70
self._page = context.pages[0] or context.new_page() # type: ignore
71
72
def _cleanup(self):
73
"""Clean up browser resources."""
74
if self._browser:
75
self._browser.close()
76
self._browser = None
77
if self._playwright:
78
self._playwright.stop()
79
self._playwright = None
80
self._page = None
81
82
def _create_session(self) -> Dict[str, str]:
83
"""Creates a new Steel browser session.
84
85
Returns:
86
Dictionary containing session details including session_id and connect_url.
87
"""
88
self._ensure_session()
89
return {
90
"session_id": self._session.id if self._session else "",
91
"connect_url": self._connect_url or "",
92
}
93
94
def navigate_to(self, url: str, connect_url: Optional[str] = None) -> str:
95
"""Navigates to a URL.
96
97
Args:
98
url (str): The URL to navigate to
99
connect_url (str, optional): The connection URL from an existing session
100
101
Returns:
102
JSON string with navigation status
103
"""
104
try:
105
self._initialize_browser(connect_url)
106
if self._page:
107
self._page.goto(url, wait_until="networkidle")
108
result = {"status": "complete", "title": self._page.title() if self._page else "", "url": url}
109
return json.dumps(result)
110
except Exception as e:
111
self._cleanup()
112
raise e
113
114
def screenshot(self, path: str, full_page: bool = True, connect_url: Optional[str] = None) -> str:
115
"""Takes a screenshot of the current page.
116
117
Args:
118
path (str): Where to save the screenshot
119
full_page (bool): Whether to capture the full page
120
connect_url (str, optional): The connection URL from an existing session
121
122
Returns:
123
JSON string confirming screenshot was saved
124
"""
125
try:
126
self._initialize_browser(connect_url)
127
if self._page:
128
self._page.screenshot(path=path, full_page=full_page)
129
return json.dumps({"status": "success", "path": path})
130
except Exception as e:
131
self._cleanup()
132
raise e
133
134
def get_page_content(self, connect_url: Optional[str] = None) -> str:
135
"""Gets the HTML content of the current page.
136
137
Args:
138
connect_url (str, optional): The connection URL from an existing session
139
140
Returns:
141
The page HTML content
142
"""
143
try:
144
self._initialize_browser(connect_url)
145
return self._page.content() if self._page else ""
146
except Exception as e:
147
self._cleanup()
148
raise e
149
150
def close_session(self) -> str:
151
"""Closes the current Steel browser session and cleans up resources.
152
153
Returns:
154
JSON string with closure status
155
"""
156
try:
157
self._cleanup()
158
159
try:
160
if self._session:
161
self.client.sessions.release(self._session.id) # type: ignore
162
except Exception as release_error:
163
logger.warning(f"Failed to release Steel session: {str(release_error)}")
164
165
self._session = None
166
self._connect_url = None
167
168
return json.dumps(
169
{
170
"status": "closed",
171
"message": "Browser resources cleaned up. Steel session released if active.",
172
}
173
)
174
except Exception as e:
175
return json.dumps({"status": "warning", "message": f"Cleanup completed with warning: {str(e)}"})
176

Step 3: Register a Steel toolkit and run an Agno Agent

Create an Agent that uses your toolkit to perform multi-step tasks.

Python
main.py
1
import os
2
from dotenv import load_dotenv
3
from agno.agent import Agent
4
from steel_tools import SteelTools
5
6
load_dotenv()
7
8
STEEL_API_KEY = os.getenv("STEEL_API_KEY") or "your-steel-api-key-here"
9
TASK = os.getenv("TASK") or "Go to https://quotes.toscrape.com and get some quotes"
10
11
def main():
12
tools = SteelTools(api_key=STEEL_API_KEY)
13
14
agent = Agent(
15
name="Web Scraper",
16
tools=[tools],
17
instructions=[
18
"Use the tools to browse and extract content.",
19
"Format results cleanly as markdown.",
20
"Always close sessions when done.",
21
],
22
markdown=True,
23
)
24
25
response = agent.run(TASK)
26
print("\nResults:\n")
27
print(response.content)
28
29
tools.close_session()
30
31
if __name__ == "__main__":
32
main()
33

Run it:

You’ll see the agent connect to a live Steel browser via CDP, navigate to the site, and extract content. A session viewer URL is printed in your Steel dashboard for live/replay views.

Complete Example

Paste the full script below into main.py and run:

Python
main.py
1
import json
2
import os
3
from typing import Any, Dict, List, Optional
4
5
from agno.tools import Toolkit
6
from agno.utils.log import log_debug, logger
7
from agno.agent import Agent
8
from playwright.sync_api import sync_playwright
9
from steel import Steel
10
11
from dotenv import load_dotenv
12
13
load_dotenv()
14
15
# Replace with your own API keys
16
STEEL_API_KEY = os.getenv("STEEL_API_KEY") or "your-steel-api-key-here"
17
18
# Replace with your own task
19
TASK = os.getenv("TASK") or "Go to https://quotes.toscrape.com and: 1. Get the first 3 quotes with authors 2. Navigate to page 2 3. Get 2 more quotes from page 2"
20
21
class SteelTools(Toolkit):
22
def __init__(
23
self,
24
api_key: Optional[str] = None,
25
**kwargs,
26
):
27
"""Initialize SteelTools.
28
29
Args:
30
api_key (str, optional): Steel API key (defaults to STEEL_API_KEY env var).
31
"""
32
self.api_key = api_key or os.getenv("STEEL_API_KEY")
33
if not self.api_key:
34
raise ValueError(
35
"STEEL_API_KEY is required. Please set the STEEL_API_KEY environment variable."
36
)
37
38
self.client = Steel(steel_api_key=self.api_key)
39
40
self._playwright = None
41
self._browser = None
42
self._page = None
43
self._session = None
44
self._connect_url = None
45
46
tools: List[Any] = []
47
tools.append(self.navigate_to)
48
tools.append(self.screenshot)
49
tools.append(self.get_page_content)
50
tools.append(self.close_session)
51
52
super().__init__(name="steel_tools", tools=tools, **kwargs)
53
54
def _ensure_session(self):
55
"""Ensures a Steel session exists, creating one if needed."""
56
if not self._session:
57
try:
58
self._session = self.client.sessions.create() # type: ignore
59
if self._session:
60
self._connect_url = f"{self._session.websocket_url}&apiKey={self.api_key}" # type: ignore
61
log_debug(f"Created new Steel session with ID: {self._session.id}")
62
except Exception as e:
63
logger.error(f"Failed to create Steel session: {str(e)}")
64
raise
65
66
def _initialize_browser(self, connect_url: Optional[str] = None):
67
"""
68
Initialize browser connection if not already initialized.
69
Use provided connect_url or ensure we have a session with a connect_url
70
"""
71
if connect_url:
72
self._connect_url = connect_url if connect_url else "" # type: ignore
73
elif not self._connect_url:
74
self._ensure_session()
75
76
if not self._playwright:
77
self._playwright = sync_playwright().start() # type: ignore
78
if self._playwright:
79
self._browser = self._playwright.chromium.connect_over_cdp(self._connect_url)
80
context = self._browser.contexts[0] if self._browser else ""
81
self._page = context.pages[0] or context.new_page() # type: ignore
82
83
def _cleanup(self):
84
"""Clean up browser resources."""
85
if self._browser:
86
self._browser.close()
87
self._browser = None
88
if self._playwright:
89
self._playwright.stop()
90
self._playwright = None
91
self._page = None
92
93
def _create_session(self) -> Dict[str, str]:
94
"""Creates a new Steel browser session.
95
96
Returns:
97
Dictionary containing session details including session_id and connect_url.
98
"""
99
self._ensure_session()
100
return {
101
"session_id": self._session.id if self._session else "",
102
"connect_url": self._connect_url or "",
103
}
104
105
def navigate_to(self, url: str, connect_url: Optional[str] = None) -> str:
106
"""Navigates to a URL.
107
108
Args:
109
url (str): The URL to navigate to
110
connect_url (str, optional): The connection URL from an existing session
111
112
Returns:
113
JSON string with navigation status
114
"""
115
try:
116
self._initialize_browser(connect_url)
117
if self._page:
118
self._page.goto(url, wait_until="networkidle")
119
result = {"status": "complete", "title": self._page.title() if self._page else "", "url": url}
120
return json.dumps(result)
121
except Exception as e:
122
self._cleanup()
123
raise e
124
125
def screenshot(self, path: str, full_page: bool = True, connect_url: Optional[str] = None) -> str:
126
"""Takes a screenshot of the current page.
127
128
Args:
129
path (str): Where to save the screenshot
130
full_page (bool): Whether to capture the full page
131
connect_url (str, optional): The connection URL from an existing session
132
133
Returns:
134
JSON string confirming screenshot was saved
135
"""
136
try:
137
self._initialize_browser(connect_url)
138
if self._page:
139
self._page.screenshot(path=path, full_page=full_page)
140
return json.dumps({"status": "success", "path": path})
141
except Exception as e:
142
self._cleanup()
143
raise e
144
145
def get_page_content(self, connect_url: Optional[str] = None) -> str:
146
"""Gets the HTML content of the current page.
147
148
Args:
149
connect_url (str, optional): The connection URL from an existing session
150
151
Returns:
152
The page HTML content
153
"""
154
try:
155
self._initialize_browser(connect_url)
156
return self._page.content() if self._page else ""
157
except Exception as e:
158
self._cleanup()
159
raise e
160
161
def close_session(self) -> str:
162
"""Closes the current Steel browser session and cleans up resources.
163
164
Returns:
165
JSON string with closure status
166
"""
167
try:
168
self._cleanup()
169
170
try:
171
if self._session:
172
self.client.sessions.release(self._session.id) # type: ignore
173
except Exception as release_error:
174
logger.warning(f"Failed to release Steel session: {str(release_error)}")
175
176
self._session = None
177
self._connect_url = None
178
179
return json.dumps(
180
{
181
"status": "closed",
182
"message": "Browser resources cleaned up. Steel session released if active.",
183
}
184
)
185
except Exception as e:
186
return json.dumps({"status": "warning", "message": f"Cleanup completed with warning: {str(e)}"})
187
188
def main():
189
print("🚀 Steel + Agno Starter")
190
print("=" * 60)
191
192
if STEEL_API_KEY == "your-steel-api-key-here":
193
print("⚠️ WARNING: Please replace 'your-steel-api-key-here' with your actual Steel API key")
194
print(" Get your API key at: https://app.steel.dev/settings/api-keys")
195
return
196
197
tools = SteelTools(api_key=STEEL_API_KEY)
198
agent = Agent(
199
name="Web Scraper",
200
tools=[tools],
201
instructions=[
202
"Extract content clearly and format nicely",
203
"Always close sessions when done",
204
],
205
markdown=True,
206
)
207
208
try:
209
response = agent.run(TASK)
210
print("\nResults:\n")
211
print(response.content)
212
except Exception as e:
213
print(f"An error occurred: {e}")
214
finally:
215
tools.close_session()
216
print("Done!")
217
218
if __name__ == "__main__":
219
main()
220

Customizing your agent’s task

Try modifying the TASK in your .env:

ENV
.env
1
# Crawl a product page and extract specs
2
TASK=Go to https://example.com/product/123 and extract the product name, price, and 5 key specs.
3
4
# Capture a screenshot-only workflow
5
TASK=Go to https://news.ycombinator.com, take a full-page screenshot, and return the page title.
6
7
# Multi-step navigation
8
TASK=Open https://docs.steel.dev, search for "session lifecycle", and summarize the key steps with anchors.

Next Steps