Quickstart

# Create project
mkdir steel-crewai-starter
cd steel-crewai-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 crewai[tools] steel-sdk python-dotenv pydantic

1
STEEL_API_KEY=your-steel-api-key-here
2
OPENAI_API_KEY=your-openai-api-key-here
3
TASK=Research AI LLMs and summarize key developments

1
import os
2
from typing import List, Optional, Type
3
from pydantic import BaseModel, Field, ConfigDict, PrivateAttr
4

5
from crewai.tools import BaseTool, EnvVar
6
from steel import Steel
7

8
class SteelScrapeWebsiteToolSchema(BaseModel):
9
    url: str = Field(description="Website URL to scrape")
10

11
class SteelScrapeWebsiteTool(BaseTool):
12
    model_config = ConfigDict(arbitrary_types_allowed=True, validate_assignment=True, frozen=False)
13
    name: str = "Steel web scrape tool"
14
    description: str = "Scrape webpages using Steel and return the contents"
15
    args_schema: Type[BaseModel] = SteelScrapeWebsiteToolSchema
16

17
    api_key: Optional[str] = None
18
    formats: Optional[List[str]] = None
19
    proxy: Optional[bool] = None
20

21
    _steel: Optional[Steel] = PrivateAttr(None)
22

23
    # For CrewAI’s packaging & env var hints
24
    package_dependencies: List[str] = ["steel-sdk"]
25
    env_vars: List[EnvVar] = [
26
        EnvVar(name="STEEL_API_KEY", description="API key for Steel services", required=True),
27
    ]
28

29
    def __init__(self, api_key: Optional[str] = None, formats: Optional[List[str]] = None,
30
                 proxy: Optional[bool] = None, **kwargs):
31
        super().__init__(**kwargs)
32
        self.api_key = api_key or os.getenv("STEEL_API_KEY")
33
        if not self.api_key:
34
            raise EnvironmentError("STEEL_API_KEY environment variable or api_key is required")
35

36
        self._steel = Steel(steel_api_key=self.api_key)
37
        self.formats = formats or ["markdown"]  # return content as Markdown by default
38
        self.proxy = proxy
39

40
    def _run(self, url: str):
41
        if not self._steel:
42
            raise RuntimeError("Steel not properly initialized")
43
        # You can set region/proxy based on your needs
44
        return self._steel.scrape(url=url, use_proxy=self.proxy, format=self.formats, region="iad")
45

1
import warnings
2
from datetime import datetime
3
from textwrap import dedent
4
from typing import List
5
from dotenv import load_dotenv
6

7
from crewai import Agent, Process, Task
8
from crewai import Crew as CrewAI
9
from crewai.agents.agent_builder.base_agent import BaseAgent
10
from crewai.project import CrewBase, agent, crew, task
11

12
warnings.filterwarnings("ignore", category=SyntaxWarning, module="pysbd")
13
load_dotenv()
14

15
TASK = os.getenv("TASK") or "Research AI LLMs and summarize key developments"
16

17
@CrewBase
18
class Crew():
19
    """Steel + CrewAI example crew"""
20
    agents: List[BaseAgent]
21
    tasks: List[Task]
22

23
    @agent
24
    def researcher(self) -> Agent:
25
        return Agent(
26
            role="Instruction-Following Web Researcher",
27
            goal="Understand and execute: {task}. Find, verify, and extract the most relevant information using the web.",
28
            backstory=(
29
                "You specialize in decomposing and executing complex instructions like '{task}', "
30
                "using web research, verification, and synthesis to produce precise, actionable findings."
31
            ),
32
            tools=[SteelScrapeWebsiteTool()],
33
            verbose=True,
34
        )
35

36
    @agent
37
    def reporting_analyst(self) -> Agent:
38
        return Agent(
39
            role="Instruction-Following Reporting Analyst",
40
            goal="Transform research outputs into a clear, complete report that fulfills: {task}",
41
            backstory=(
42
                "You convert research into exhaustive, well-structured reports that directly address "
43
                "the original instruction '{task}', ensuring completeness and clarity."
44
            ),
45
            tools=[SteelScrapeWebsiteTool()],
46
            verbose=True,
47
        )
48

49
    @task
50
    def research_task(self) -> Task:
51
        return Task(
52
            description=dedent("""
53
                Interpret and execute the following instruction: {task}
54
                Use the web as needed. Cite and include key sources.
55
                Consider the current year: {current_year}.
56
            """),
57
            expected_output="A structured set of findings and sources that directly satisfy the instruction: {task}",
58
            agent=self.researcher(),
59
        )
60

61
    @task
62
    def reporting_task(self) -> Task:
63
        return Task(
64
            description=dedent("""
65
                Review the research context and produce a complete report that fulfills the instruction.
66
                Ensure completeness, accuracy, and clear structure. Include citations.
67
            """),
68
            expected_output=(
69
                "A comprehensive markdown report that satisfies the instruction: {task}. "
70
                "Formatted as markdown without '```'"
71
            ),
72
            agent=self.reporting_analyst(),
73
        )
74

75
    @crew
76
    def crew(self) -> CrewAI:
77
        """Creates the sequential crew pipeline"""
78
        return CrewAI(
79
            agents=self.agents,
80
            tasks=self.tasks,
81
            process=Process.sequential,
82
            verbose=True,
83
        )
84

1
def main():
2
    print("🚀 Steel + CrewAI Starter")
3
    print("=" * 60)
4

5
    if not os.getenv("STEEL_API_KEY") or os.getenv("STEEL_API_KEY") == "your-steel-api-key-here":
6
        print("⚠️  WARNING: Please set STEEL_API_KEY in your .env")
7
        print("   Get your key at: https://app.steel.dev/settings/api-keys")
8
        return
9

10
    inputs = {
11
        "task": TASK,
12
        "current_year": str(datetime.now().year),
13
    }
14

15
    try:
16
        print("Running crew...")
17
        Crew().crew().kickoff(inputs=inputs)
18
        print("\n✅ Done. (If your task wrote to a file, check your project folder.)")
19
    except Exception as e:
20
        print(f"❌ Error while running the crew: {e}")
21

22
if __name__ == "__main__":
23
    main()
24

1
import os
2
import warnings
3
from datetime import datetime
4
from textwrap import dedent
5
from typing import List, Optional, Type
6

7
from crewai import Agent, Process, Task
8
from crewai import Crew as CrewAI
9
from crewai.agents.agent_builder.base_agent import BaseAgent
10
from crewai.project import CrewBase, agent, crew, task
11
from crewai.tools import BaseTool, EnvVar
12
from dotenv import load_dotenv
13
from pydantic import BaseModel, ConfigDict, Field, PrivateAttr
14
from steel import Steel
15

16
warnings.filterwarnings("ignore", category=SyntaxWarning, module="pysbd")
17
load_dotenv()
18

19
# Replace with your own API keys
20
STEEL_API_KEY = os.getenv('STEEL_API_KEY') or "your-steel-api-key-here"
21

22
# Replace with your own task
23
TASK = os.getenv('TASK') or 'Research AI LLMs and summarize key developments'
24

25
class SteelScrapeWebsiteToolSchema(BaseModel):
26
    url: str = Field(description="Website URL")
27

28
class SteelScrapeWebsiteTool(BaseTool):
29
    model_config = ConfigDict(arbitrary_types_allowed=True, validate_assignment=True, frozen=False)
30
    name: str = "Steel web scrape tool"
31
    description: str = "Scrape webpages using Steel and return the contents"
32
    args_schema: Type[BaseModel] = SteelScrapeWebsiteToolSchema
33
    api_key: Optional[str] = None
34
    formats: Optional[List[str]] = None
35
    proxy: Optional[bool] = None
36

37
    _steel: Optional[Steel] = PrivateAttr(None)
38
    package_dependencies: List[str] = ["steel-sdk"]
39
    env_vars: List[EnvVar] = [
40
        EnvVar(name="STEEL_API_KEY", description="API key for Steel services", required=True),
41
    ]
42

43
    def __init__(self, api_key: Optional[str] = None, formats: Optional[List[str]] = None,
44
                 proxy: Optional[bool] = None, **kwargs):
45
        super().__init__(**kwargs)
46
        self.api_key = api_key or os.getenv("STEEL_API_KEY")
47
        if not self.api_key:
48
            raise EnvironmentError("STEEL_API_KEY environment variable or api_key is required")
49

50
        self._steel = Steel(steel_api_key=self.api_key)
51
        self.formats = formats or ["markdown"]
52
        self.proxy = proxy
53

54
    def _run(self, url: str):
55
        if not self._steel:
56
            raise RuntimeError("Steel not properly initialized")
57
        return self._steel.scrape(url=url, use_proxy=self.proxy, format=self.formats, region="iad")
58

59
@CrewBase
60
class Crew():
61
    """Crew crew"""
62
    agents: List[BaseAgent]
63
    tasks: List[Task]
64

65
    @agent
66
    def researcher(self) -> Agent:
67
        return Agent(
68
            role="Instruction-Following Web Researcher",
69
            goal="Understand and execute: {task}. Find, verify, and extract the most relevant information using the web.",
70
            backstory=(
71
                "You specialize in decomposing and executing complex instructions like '{task}', "
72
                "using web research, verification, and synthesis to produce precise, actionable findings."
73
            ),
74
            tools=[SteelScrapeWebsiteTool()],
75
            verbose=True
76
        )
77

78
    @agent
79
    def reporting_analyst(self) -> Agent:
80
        return Agent(
81
            role="Instruction-Following Reporting Analyst",
82
            goal="Transform research outputs into a clear, complete report that fulfills: {task}",
83
            backstory=(
84
                "You convert research into exhaustive, well-structured reports that directly address "
85
                "the original instruction '{task}', ensuring completeness and clarity."
86
            ),
87
            tools=[SteelScrapeWebsiteTool()],
88
            verbose=True
89
        )
90

91
    @task
92
    def research_task(self) -> Task:
93
        return Task(
94
            description=dedent("""
95
                Interpret and execute the following instruction: {task}
96
                Use the web as needed. Cite and include key sources.
97
                Consider the current year: {current_year}.
98
            """),
99
            expected_output="A structured set of findings and sources that directly satisfy the instruction: {task}",
100
            agent=self.researcher()
101
        )
102

103
    @task
104
    def reporting_task(self) -> Task:
105
        return Task(
106
            description=dedent("""
107
                Review the research context and produce a complete report that fulfills the instruction.
108
                Ensure completeness, accuracy, and clear structure. Include citations.
109
            """),
110
            expected_output="A comprehensive markdown report that satisfies the instruction: {task}. Formatted as markdown without '```'",
111
            agent=self.reporting_analyst(),
112
        )
113

114
    @crew
115
    def crew(self) -> CrewAI:
116
        """Creates the Crew crew"""
117
        return CrewAI(
118
            agents=self.agents,
119
            tasks=self.tasks,
120
            process=Process.sequential,
121
            verbose=True,
122
        )
123

124
def main():
125
    print("🚀 Steel + CrewAI Starter")
126
    print("=" * 60)
127

128
    if STEEL_API_KEY == "your-steel-api-key-here":
129
        print("⚠️  WARNING: Please replace 'your-steel-api-key-here' with your actual Steel API key")
130
        print("   Get your API key at: https://app.steel.dev/settings/api-keys")
131
        return
132

133
    inputs = {
134
        'task': TASK,
135
        'current_year': str(datetime.now().year)
136
    }
137

138
    try:
139
        print("Running crew...")
140
        Crew().crew().kickoff(inputs=inputs)
141
        print("\n✅ Crew finished.")
142
    except Exception as e:
143
        print(f"❌ An error occurred while running the crew: {e}")
144

145
if __name__ == "__main__":
146
    main()
147

1
TASK = "Visit https://docs.steel.dev and summarize the Sessions API lifecycle with citations."
2
# or
3
TASK = "Find the latest research trends in open-weights LLMs and produce a bullet summary with 5 sources."
4
# or
5
TASK = "Compare two AI agent frameworks and write a short pros/cons table with links."