> ## Documentation Index
> Fetch the complete documentation index at: https://docs.agno.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Run and Stream

> The stream is parsed into typed Agno events so applications can handle content, tools, completion, and errors explicitly.

```python run_and_stream.py theme={null}
"""Run an agent through AgentOS with normal and streaming responses.

The stream is parsed into typed Agno events so applications can handle
content, tools, completion, and errors explicitly.

Prerequisites: start ``_server.py`` and set OPENAI_API_KEY.
Run: .venvs/demo/bin/python cookbook/05_agent_os/03_python_client/02_run_and_stream.py
Try: watch the calculator tool events arrive before the final content.
"""

import asyncio

from agno.client import AgentOSClient
from agno.run.agent import (
    RunCompletedEvent,
    RunContentEvent,
    RunErrorEvent,
    RunStartedEvent,
    ToolCallCompletedEvent,
    ToolCallStartedEvent,
)

BASE_URL = "http://localhost:7778"
AGENT_ID = "assistant"


# ---------------------------------------------------------------------------
# Create the Client
# ---------------------------------------------------------------------------


async def cancel_active_run(client: AgentOSClient, run_id: str) -> None:
    """Cancel an active agent run."""
    await client.cancel_agent_run(AGENT_ID, run_id)


async def run_examples() -> None:
    """Run one complete response and one typed event stream."""
    client = AgentOSClient(base_url=BASE_URL)

    response = await client.run_agent(
        agent_id=AGENT_ID,
        message="What is 17 multiplied by 23? Use the calculator.",
    )
    print(f"Non-streaming run: {response.run_id}")
    print(response.content)

    print("\nStreaming response:")
    async for event in client.run_agent_stream(
        agent_id=AGENT_ID,
        message="Use the calculator to add 41 and 1, then explain the result.",
    ):
        if isinstance(event, RunStartedEvent):
            print(f"Run started: {event.run_id}")
        elif isinstance(event, ToolCallStartedEvent):
            tool_name = event.tool.tool_name if event.tool else "unknown"
            print(f"\nTool started: {tool_name}")
        elif isinstance(event, ToolCallCompletedEvent):
            tool_name = event.tool.tool_name if event.tool else "unknown"
            print(f"Tool completed: {tool_name}")
        elif isinstance(event, RunContentEvent) and event.content is not None:
            print(event.content, end="", flush=True)
        elif isinstance(event, RunCompletedEvent):
            print(f"\nRun completed: {event.run_id}")
        elif isinstance(event, RunErrorEvent):
            raise RuntimeError(event.content or "Agent run failed")

    print(
        "\nTeams and workflows use the same pattern through "
        "run_team/run_team_stream and run_workflow/run_workflow_stream."
    )
    print("Cancel an active run with await client.cancel_agent_run(agent_id, run_id).")


# ---------------------------------------------------------------------------
# Run the Example
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    asyncio.run(run_examples())
```

## Run the Example

<Steps>
  <Snippet file="create-venv-step.mdx" />

  <Step title="Install dependencies">
    ```bash theme={null}
    uv pip install -U "agno[os]" chromadb openai
    ```
  </Step>

  <Step title="Export your API keys">
    <CodeGroup>
      ```bash Mac/Linux theme={null}
      export OPENAI_API_KEY="your_openai_api_key_here"
      ```

      ```bash Windows theme={null}
      $Env:OPENAI_API_KEY="your_openai_api_key_here"
      ```
    </CodeGroup>
  </Step>

  <Step title="Clone Agno">
    Clone the pinned Agno source and run the remaining commands from its root:

    ```bash theme={null}
    git clone https://github.com/agno-agi/agno.git
    cd agno
    git checkout v3.0.4
    ```
  </Step>

  <Step title="Start the AgentOS server">
    Open another terminal in the parent directory where `.venv` and the cloned `agno` directory are siblings, then start the shared server on port 7778:

    <CodeGroup>
      ```bash Mac/Linux theme={null}
      source .venv/bin/activate
      export OPENAI_API_KEY="your_openai_api_key_here"
      cd agno
      python cookbook/05_agent_os/03_python_client/_server.py
      ```

      ```powershell Windows theme={null}
      .venv\Scripts\activate
      $Env:OPENAI_API_KEY="your_openai_api_key_here"
      Set-Location agno
      python cookbook/05_agent_os/03_python_client/_server.py
      ```
    </CodeGroup>
  </Step>

  <Step title="Run the example">
    Run the example from the repository root:

    ```bash theme={null}
    python cookbook/05_agent_os/03_python_client/02_run_and_stream.py
    ```
  </Step>
</Steps>

Full source: [cookbook/05\_agent\_os/03\_python\_client/02\_run\_and\_stream.py](https://github.com/agno-agi/agno/blob/v3.0.4/cookbook/05_agent_os/03_python_client/02_run_and_stream.py)