mcp-use__mcp-use/docs/python/agent/interactive-chat-patterns.mdx

---
title: "Interactive Chat Patterns"
description: "Create interactive chat interfaces with persistent conversation memory"
icon: "message-circle"
---

## Building a chat loop

With mcp-use you can build interactive interface where users can have conversations with
your `MCPAgent`, maintaining context and memory across multiple queries.

## Basic chat loop
Here's a basic chat-loop with conversation memory enabled:

<CodeGroup>
```python Python
import asyncio
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from mcp_use import MCPAgent, MCPClient

async def basic_chat_loop():
    """Simple console chat loop with MCPAgent"""
    # Load environment variables
    load_dotenv()

    # MCP server configuration
    config = {
        "mcpServers": {
            "playwright": {
                "command": "npx",
                "args": ["@playwright/mcp@latest"],
                "env": {"DISPLAY": ":1"}
            },
            "filesystem": {
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
            }
        }
    }

    # Create client and agent
    client = MCPClient(config)
    llm = ChatOpenAI(model="gpt-5.5")

    agent = MCPAgent(llm=llm,
        client=client,
        memory_enabled=True, # Enable memory to track conversation history
        max_steps=20)

    # Some initial messages
    print("🤖 MCP Agent Chat")
    print("Type 'quit/exit' to exit the chat.")
    print("Type 'clear' to clear conversation history")

    try:
        while True:
            user_input = input("\nYou: ")

            if user_input.lower() in ['quit', 'exit']:
                print("👋 Goodbye!")
                break

            if user_input.lower() == 'clear':
                agent.clear_conversation_history()
                print("🧹 Conversation history cleared.")
                continue

            # Skip empty messages
            if not user_input:
                continue

            try:
                print("\n🤖 Assistant: ", end="", flush=True)
                response = await agent.run(user_input)
                print(response)
            except KeyboardInterrupt: # Handle keyboard interrupt
                print("\n\n⏸️ Interrupted by user")
                break
            except Exception as e:
                print(f"\n❌ Error: {e}")
                print("Please try again or type 'exit' to quit.")
    finally:
        await client.close_all_sessions()

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

## Streaming Chat Loop

Here's a chat loop with streaming responses enabled:

<CodeGroup>
```python Python
import asyncio
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from mcp_use import MCPAgent, MCPClient

async def streaming_chat_loop():
    """Chat loop with streaming responses with MCPAgent"""
    # Load environment variables
    load_dotenv()

    # MCP server configuration
    config = {
        "mcpServers": {
            "playwright": {
                "command": "npx",
                "args": ["@playwright/mcp@latest"],
                "env": {"DISPLAY": ":1"}
            }
        }
    }

    # Create client and agent
    client = MCPClient(config)
    llm = ChatOpenAI(model="gpt-5.5")

    agent = MCPAgent(llm=llm,
        client=client,
        memory_enabled=True, # Enable memory to track conversation history
        max_steps=20)

    # Some initial messages
    print("🤖 MCP Agent Chat (Streaming)")
    print("Type 'quit/exit' to exit the chat.")
    print("Type 'clear' to clear conversation history")

    try:
        while True:
            user_input = input("\nYou: ")

            if user_input.lower() in ['quit', 'exit']:
                print("👋 Goodbye!")
                break

            if user_input.lower() == 'clear':
                agent.clear_conversation_history()
                print("🧹 Conversation history cleared.")
                continue

            if not user_input: # Skip empty messages
                continue

            try:
                print("\n🤖 Assistant: ", end="", flush=True)

                # Stream the response
                async for chunk in agent.stream(user_input):
                    print(chunk, end="", flush=True)
                print()
            except KeyboardInterrupt: # Handle keyboard interrupt
                print("\n\n⏸️ Interrupted by user")
                break
            except Exception as e:
                print(f"\n❌ Error: {e}")
                print("Please try again or type 'exit' to quit.")
    finally:
        await client.close_all_sessions()

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

## Chat Loop with Structured I/O

It's possible to create a chat loop that can handle both natural language and structured inputs, allowing users to request specific tasks or analyses in a structured format. Here's an example of how to implement this:

<CodeGroup>
```python Python
import asyncio
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from mcp_use import MCPAgent, MCPClient
from typing import Optional

class TaskRequest(BaseModel):
    task_type: Optional[str] = Field(description="The type of task to perform")
    description: Optional[str] = Field(description="Detailed description of the task")
    priority: Optional[str] = Field(description="Priority level: low, medium, high")

async def structured_chat_loop():
    """Chat loop that can handle both natural language and structured inputs."""
    # Load environment variables
    load_dotenv()

    # MCP server configuration
    config = {
        "mcpServers": {
            "playwright": {
                "command": "npx",
                "args": ["@playwright/mcp@latest"],
                "env": {"DISPLAY": ":1"}
            }
        }
    }

    # Create client and agent
    client = MCPClient(config)
    llm = ChatOpenAI(model="gpt-5.5")

    agent = MCPAgent(
        llm=llm,
        client=client,
        memory_enabled=True, # Enable memory to track conversation history
        max_steps=20
    )

    # Initial messages
    print("🤖 MCP Agent Chat (Structured)")
    print("You can chat naturally or request structured task analysis")
    print("Type 'task' to create a structured task request")

    try:
        while True:
            user_input = input("\nYou: ")
            if user_input.lower() in ['exit', 'quit']:
                print("👋 Goodbye!")
                break

            try:
                if user_input.lower() == 'task':
                    print("\n📋 Creating structured task...")
                    task_description = input("Describe your task: ")

                    task: TaskRequest = await agent.run(
                        f"Analyze a task with the following description: {task_description}",
                        output_schema=TaskRequest
                    )

                    # Print task analysis
                    print(f"\n✅ Task Analysis:")
                    print(f"• Type: {task.task_type}")
                    print(f"• Description: {task.description}")
                    print(f"• Priority: {task.priority or 'low'}")

                    proceed = input("\nDo you want to proceed with this task? (y/n)")
                    if proceed.lower() == 'y':
                        response = await agent.run(
                            f"Execute the following task: {task.description}"
                        )
                        print(f"\n🤖 Assistant: {response}")
                else:
                    # Regular conversation
                    response = await agent.run(user_input)
                    print(f"\n🤖 Assistant: {response}")
            except KeyboardInterrupt:
                print("\n👋 Goodbye!")
                break
            except Exception as e:
                print(f"❌ Error: {e}")
                print("Please try again or type 'exit' to quit.")

    finally:
        await client.close_all_sessions()

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

## Next Steps

<CardGroup cols={3}>
  <Card title="Agent Configuration" icon="cloud" 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="Security Best Practices" icon="shield" href="/python/development/security">
    Learn how to secure your MCP deployments
  </Card>
</CardGroup>