# Tutorial: Build your first agent

> Create an Everruns agent in Python, send a message, and stream the response.

Source: <https://everruns.com/docs/tutorials/building-agents-using-sdk/>

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](https://everruns.com/docs/how-to/).

If you prefer a Jupyter notebook walkthrough, start with [Run an Agent](https://everruns.com/docs/tutorials/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

* An Everruns account. The quickest option is [Everruns Cloud](https://app.everruns.com), which includes a built-in model provider and starter credit. A [self-hosted Platform](https://everruns.com/docs/getting-started/docker-compose/) works too.
* A personal access token: **Settings** > **Personal Access Tokens** in the UI.
* Python 3.10 or newer.

```bash
pip install everruns-sdk
export EVERRUNS_API_KEY=evr_pat_...
export EVERRUNS_API_URL=https://app.everruns.com/api  # or http://localhost:9300/api when self-hosted
```

## Step 1: Connect to the server

The `Everruns` client reads `EVERRUNS_API_KEY` and `EVERRUNS_API_URL` from the environment.

```python
import asyncio
from everruns_sdk import AgentCapabilityConfig, Everruns


client = Everruns()
```

## Step 2: Create an agent

An **agent** is the configuration: a name, a system prompt, and a set of capabilities (tools). `apply_by_name` creates the agent, or updates it when you run the script again. Names are lowercase words joined by hyphens.

```python
async def main():
    agent = await client.agents.apply_by_name(
        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=[
            AgentCapabilityConfig(ref="web_fetch"),
            AgentCapabilityConfig(ref="session_file_system"),
            AgentCapabilityConfig(ref="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.

```python
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 starts a durable turn that runs the agent’s reason-act loop. The call returns as soon as the message is recorded; the response arrives as events.

```python
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 and handles SSE reconnection and heartbeats for you. A stream only delivers events written after it connects, and a fast model can start answering before that. So start the stream from the event that recorded your message (`input.message`): nothing after it is missed.

```python
latest_input = await client.events.list(
    session.id, types=["input.message"], order_desc=True, limit=1
)


async for event in client.events.stream(session.id, since_id=latest_input[0].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 answer 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:

```python
import asyncio
from everruns_sdk import AgentCapabilityConfig, Everruns




async def main():
    client = Everruns()


    agent = await client.agents.apply_by_name(
        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=[
            AgentCapabilityConfig(ref="web_fetch"),
            AgentCapabilityConfig(ref="session_file_system"),
            AgentCapabilityConfig(ref="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?",
    )


    latest_input = await client.events.list(
        session.id, types=["input.message"], order_desc=True, limit=1
    )


    async for event in client.events.stream(session.id, since_id=latest_input[0].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:

```bash
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](https://everruns.com/docs/explanation/agentic-loop/) for the design.

## Next steps

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

* [Equip an agent with tools](https://everruns.com/docs/how-to/equip-agents-with-tools/): explore the full capability catalog.
* [Define agents as files](https://everruns.com/docs/how-to/define-agents-as-files/): version-control your agent definitions.
* [Stream events with the SDK](https://everruns.com/docs/how-to/stream-events/): richer streaming patterns.
* [Handle errors and cancel turns](https://everruns.com/docs/how-to/handle-errors-and-cancellation/): what to do when things go wrong.
* [Orchestrate multi-agent pipelines](https://everruns.com/docs/how-to/orchestrate-multi-agent-pipelines/): chain agents together.
* [Publish an agent as a Slack app](https://everruns.com/docs/how-to/publish-to-slack/): deploy to a channel.

For background, read [Core concepts](https://everruns.com/docs/getting-started/concepts/) and [The agentic loop](https://everruns.com/docs/explanation/agentic-loop/).