class MyAgent(BaseAgent):
@staticmethod
def name() -> str:
return "my-agent"
def version(self) -> str | None:
return "1.0.0"
async def setup(self, environment: BaseEnvironment) -> None:
"""Called once before the agent starts.
Use this to discover available tools, configure state,
or perform any one-time initialization.
"""
self.tools = await environment.list_tools()
async def run(
self,
instruction: str,
environment: BaseEnvironment,
context: RunArtifacts,
) -> None:
"""Execute the agent's task.
Populate the context (RunArtifacts) as execution progresses
so that partial results are captured even on timeout or error.
"""
context.record_message({"role": "user", "content": instruction})
# Your agent logic here:
# 1. Read the instruction
# 2. Decide which tools to call
# 3. Call tools via environment.call_tool(server, name, params)
# 4. Record each tool call and result in context
# 5. Iterate until done
call = ToolCall(tool_server="email-env", tool_name="send_email", parameters={...})
result = await environment.call_tool(call.tool_server, call.tool_name, call.parameters)
context.record_tool_call(call, result)