Skip to content

Tutorial: Build your first agent

By the end of this tutorial you will have created an Everruns agent, started a session, sent a message, and streamed the response — using the official Python SDK.

This is a tutorial: a guided lesson. We make all the choices for you. When you want to do something different (different language, different tools, different patterns), follow up with the How-to guides.

If you prefer a Jupyter notebook walkthrough, start with Run an Agent instead.

What you’ll build

A research-assistant agent with web access. You’ll send it a topic and watch it answer.

What you need

  • A running Everruns instance. The easiest path is the Docker Compose quickstart.
  • Python 3.10 or newer.
  • An LLM provider configured (an OpenAI or Anthropic API key set in the Everruns UI).
Terminal window
pip install everruns-sdk

Step 1 — Connect to the server

The Everruns client reads EVERRUNS_API_KEY and EVERRUNS_API_URL from the environment. For a local just start-dev deployment, API key "dev" works.

import asyncio
from everruns_sdk import Everruns
client = Everruns(api_key="dev", base_url="http://localhost:9300/api")

Step 2 — Create an agent

An agent is the configuration: a name, a system prompt, and a set of capabilities (tools).

async def main():
agent = await client.agents.create(
name="Research Assistant",
system_prompt=(
"You are a research assistant. When given a topic, you:\n"
"1. Fetch relevant web pages\n"
"2. Save your notes to /workspace\n"
"3. Produce a concise summary"
),
capabilities=["web_fetch", "session_file_system", "current_time"],
)
print(f"Agent: {agent.id}")
asyncio.run(main())

The capabilities give the agent its tools: web_fetch to retrieve URLs, session_file_system for an isolated workspace, current_time to know what day it is.

Step 3 — Start a session

A session is a working conversation with the agent. It owns the conversation history, an isolated virtual filesystem, and key/value storage.

session = await client.sessions.create(
agent_id=agent.id,
title="Research: Durable Execution",
)
print(f"Session: {session.id}")

Step 4 — Send a message

Sending a user message queues a durable workflow that runs the agent’s reason–act loop. The call returns immediately — the response arrives as events.

await client.messages.create(
session.id,
"Research durable execution engines. What are the main approaches?"
)

Step 5 — Stream the response

client.events.stream(...) returns an async iterator over typed events. It handles SSE reconnection, heartbeats, and resumption for you.

async for event in client.events.stream(session.id):
if event.type == "output.message.delta":
print(event.data.get("delta", ""), end="", flush=True)
elif event.type == "tool.started":
tool = event.data.get("tool_call", {}).get("name", "")
print(f"\n [tool] {tool}")
elif event.type == "turn.completed":
print("\n[done]")
break
elif event.type == "turn.failed":
print(f"\n[failed: {event.data.get('error')}]")
break

You’ll see the agent’s reasoning stream token-by-token, with [tool] markers each time it fetches a URL or writes to its workspace.

Step 6 — Put it together

Here’s the complete program:

import asyncio
from everruns_sdk import Everruns
async def main():
client = Everruns(api_key="dev", base_url="http://localhost:9300/api")
agent = await client.agents.create(
name="Research Assistant",
system_prompt=(
"You are a research assistant. When given a topic, you:\n"
"1. Fetch relevant web pages\n"
"2. Save your notes to /workspace\n"
"3. Produce a concise summary"
),
capabilities=["web_fetch", "session_file_system", "current_time"],
)
session = await client.sessions.create(
agent_id=agent.id,
title="Research: Durable Execution",
)
await client.messages.create(
session.id,
"Research durable execution engines. What are the main approaches?",
)
async for event in client.events.stream(session.id):
if event.type == "output.message.delta":
print(event.data.get("delta", ""), end="", flush=True)
elif event.type == "tool.started":
tool = event.data.get("tool_call", {}).get("name", "")
print(f"\n [tool] {tool}")
elif event.type == "turn.completed":
print("\n[done]")
break
elif event.type == "turn.failed":
print(f"\n[failed: {event.data.get('error')}]")
break
await client.close()
asyncio.run(main())

Save as tutorial.py and run:

Terminal window
python tutorial.py

You should see the agent stream a research response, fetching a few URLs along the way.

What just happened

You configured an agent (long-lived), started a session (per-conversation), and consumed the event stream (per-turn). Those three layers — configuration, runtime, data — are the spine of every Everruns application.

The agent loop you watched run is the reason–act cycle. The model reasons (returns text or tool calls), tools execute, results feed back in, repeat until the model produces a final answer. See The agentic loop for the design.

Next steps

Common follow-ups, each as a focused how-to:

For background, read Core concepts and The agentic loop.