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

# Streaming Agent Output

> Implement real-time streaming of agent actions and responses

mcp-use supports multiple approaches for streaming agent output, allowing you to receive incremental results, tool actions, and intermediate steps as they are generated by the agent.

## Step-by-Step Streaming

The `stream` method provides a clean interface for receiving intermediate steps during agent execution. Each step represents a tool call and its result.

<CodeGroup>
  ```python Python theme={null}
  import asyncio

  from dotenv import load_dotenv
  from langchain_anthropic import ChatAnthropic

  from mcp_use import MCPAgent, MCPClient


  async def main():
      """Run the example using a configuration file."""
      # Load environment variables
      load_dotenv()

      config = {
          "mcpServers": {"playwright": {"command": "npx", "args": ["@playwright/mcp@latest"], "env": {"DISPLAY": ":1"}}}
      }
      # Create MCPClient from config file
      client = MCPClient(config=config)
      # Create LLM
      llm = ChatAnthropic(model="claude-sonnet-4-5")
      # Create agent with the client
      agent = MCPAgent(llm=llm, client=client, max_steps=30)
      # Run the query
      async for step in agent.stream(
          """
          Can you go on github and tell me how many stars the mcp-use project has?
          """,
          max_steps=30,
      ):
          if isinstance(step, str):
              print("-------------Result--------------------------")
              print("Result:", step)
          else:
              action, observation = step
              print("-------------Log--------------------------")
              print("Log:", action.log)
              print("--------------------------------")
              print("-------------Calling--------------------------")
              print("Calling:", action.tool)
              print("--------------------------------")
              print("-------------Input--------------------------")
              print("Input:", action.tool_input)
              print("--------------------------------")
              print("-------------Observation--------------------------")
              print("Observation:", observation)
              print("--------------------------------")


  if __name__ == "__main__":
      # Run the appropriate example
      asyncio.run(main())
  ```
</CodeGroup>

### Understanding Yielded Objects

The `stream` method yields two types of objects during execution:

<CardGroup cols={1}>
  <Card title="Intermediate Steps (Tuple)" icon="link">
    Each tool call yields `(AgentAction, observation)`:

    * **`action.log`** - The agent's reasoning text explaining why it's calling the tool
    * **`action.tool`** - The name of the tool being called (e.g., "add", "echo")
    * **`action.tool_input`** - The input arguments passed to the tool as a dictionary
    * **`observation`** - The result returned by the tool after execution
  </Card>

  <Card title="Final Result (String)" icon="check">
    After all tool calls complete, the agent yields a final string response with the answer to the query.
  </Card>
</CardGroup>

## Low-Level Event Streaming

For more granular control, use the `stream_events` method to get real-time output events:

<CodeGroup>
  ```python Python theme={null}
  import asyncio
  from langchain_openai import ChatOpenAI
  from mcp_use import MCPAgent, MCPClient

  async def basic_streaming_example():
      # Setup agent
      config = {
          "mcpServers": {
              "playwright": {
                  "command": "npx",
                  "args": ["@playwright/mcp@latest"]
              }
          }
      }

      client = MCPClient(config)
      llm = ChatOpenAI(model="gpt-4")
      agent = MCPAgent(llm=llm, client=client)

      # Stream the agent's response
      print("Agent is working...")
      async for chunk in agent.stream_events("Search for the latest Python news and summarize it"):
          print(chunk, end="", flush=True)

      print("\n\nDone!")

  if __name__ == "__main__":
      asyncio.run(basic_streaming_example())
  ```
</CodeGroup>

<Tip>
  The streaming API is based on LangChain's `stream_events` method. For more details on event types and data structure, check the [LangChain streaming documentation](https://python.langchain.com/docs/how_to/streaming/).
</Tip>

## Choosing the Right Streaming Method

<CardGroup cols={2}>
  <Card title="Use stream() when:" icon="list-ordered">
    • You want to show step-by-step progress
    • You need to process each tool call individually
    • You're building a workflow UI
    • You want simple, clean step tracking
  </Card>

  <Card title="Use stream_events() when:" icon="code">
    • You need fine-grained control over events
    • You're building real-time chat interfaces
    • You want to stream LLM reasoning text
    • You need custom event filtering
  </Card>
</CardGroup>

## Examples

### Building a Streaming UI

Here's an example of how you might build a simple console UI for streaming:

```python streaming_ui.py theme={null}
import asyncio
import sys
from datetime import datetime
from langchain_openai import ChatOpenAI
from mcp_use import MCPAgent, MCPClient


class StreamingUI:
    def __init__(self):
        self.current_thought = ""
        self.tool_outputs = []
        self.final_answer = ""

    def clear_line(self):
        """Clear the current line in terminal"""
        sys.stdout.write("\r\033[K")

    def print_status(self, status, tool=None):
        """Print colored status updates"""
        timestamp = datetime.now().strftime("%H:%M:%S")
        if tool:
            print(f"\033[94m[{timestamp}] {status}: {tool}\033[0m")
        else:
            print(f"\033[92m[{timestamp}] {status}\033[0m")

    def print_thinking(self, text):
        """Print agent's reasoning in real-time"""
        self.clear_line()
        truncated = text[:80] + "..." if len(text) > 80 else text
        sys.stdout.write(f"\033[93m💭 Thinking: {truncated}\033[0m")
        sys.stdout.flush()

    def print_tool_result(self, tool_name, result):
        """Print tool execution results"""
        print(f"\n\033[96m🔧 {tool_name} result:\033[0m")
        # Truncate long results
        display_result = result[:200] + "..." if len(result) > 200 else result
        print(f"   {display_result}")


async def streaming_ui_example():
    config = {"mcpServers": {"playwright": {"command": "npx", "args": ["@playwright/mcp@latest"]}}}

    client = MCPClient(config)
    llm = ChatOpenAI(model="gpt-4", streaming=True)
    agent = MCPAgent(llm=llm, client=client)

    ui = StreamingUI()

    query = "What are the current trending topics on Hacker News?"

    print("🤖 MCP Agent - Interactive Session")
    print("=" * 50)
    print(f"Query: {query}")
    print("=" * 50)

    current_tool = None
    current_reasoning = ""

    async for event in agent.stream_events(query):
        event_type = event.get("event")
        data = event.get("data", {})

        if event_type == "on_chat_model_start":
            ui.print_status("Starting to plan")

        elif event_type == "on_chat_model_stream":
            chunk = data.get("chunk", {})
            if hasattr(chunk, "content") and chunk.content:
                current_reasoning += chunk.content
                ui.print_thinking(current_reasoning)

        elif event_type == "on_tool_start":
            current_tool = data.get("input", {}).get("tool_name")
            if current_tool:
                print("\n")  # New line after thinking
                ui.print_status("Executing tool", current_tool)
                current_reasoning = ""  # Reset for next iteration

        elif event_type == "on_tool_end":
            output = data.get("output")
            if current_tool and output:
                ui.print_tool_result(current_tool, str(output))

        elif event_type == "on_chain_end":
            print("\n")
            ui.print_status("Task completed!")

            # Extract final answer
            final_output = data.get("output")
            if final_output:
                print(f"\n\033[92m📋 Final Answer:\033[0m")
                print(f"{final_output}")


if __name__ == "__main__":
    asyncio.run(streaming_ui_example())

```

### Web Streaming with FastAPI

For web applications, you can stream agent output using Server-Sent Events:

```python web_streaming.py theme={null}
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import json
from langchain_openai import ChatOpenAI
from mcp_use import MCPAgent, MCPClient
import time
app = FastAPI()

async def create_agent():
    config = {
        "mcpServers": {
            "playwright": {
                "command": "npx",
                "args": ["@playwright/mcp@latest"]
            }
        }
    }
    client = MCPClient(config)
    llm = ChatOpenAI(model="gpt-4", streaming=True)
    return MCPAgent(llm=llm, client=client)

@app.get("/stream/{query}")
async def stream_agent_response(query: str):
    """Stream agent response using Server-Sent Events"""

    async def event_generator():
        agent = await create_agent()

        async for event in agent.stream_events(query):
            event_type = event.get("event")
            if event_type == "on_chat_model_stream":
                data = event["data"]["chunk"].content
            else:
                data = ""
            # Format as SSE
            sse_data = {
                "type": event_type,
                "timestamp": time.time(),
                "data": data
            }

            yield f"data: {json.dumps(sse_data)}\n\n"

        yield "data: [DONE]\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/plain",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        }
    )

@app.get("/")
async def root():
    return {"message": "MCP Agent Streaming API"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
```

## Next Steps

<CardGroup cols={3}>
  <Card title="Agent Configuration" icon="gear" href="/python/agent/agent-configuration">
    Learn more about configuring agents for optimal streaming performance
  </Card>

  <Card title="Multi-Server Setup" icon="server" href="/python/client/multi-server-setup">
    Stream output from agents using multiple MCP servers
  </Card>

  <Card title="Agent Configuration" icon="zap" href="/python/agent/agent-configuration">
    Learn about configuring agents for optimal performance
  </Card>
</CardGroup>