import deepagents

print(dir(deepagents))

!python --version

# 安装deepagents依赖
!pip install deepagents

# 查看版本
!pip list | grep -E 'langchain|deepagents'

# from langchain_openai import ChatOpenAI
from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv

# 1. 加载.env环境变量
load_dotenv(override=True)

# 2. 初始化模型
model = ChatDeepSeek(model="deepseek-chat", temperature=0)
model.invoke("你好")

from deepagents import create_deep_agent

create_deep_agent?

# 安装网络搜索工具
!pip install langchain-tavily

from deepagents import create_deep_agent
from langchain_tavily import TavilySearch
from langgraph.checkpoint.memory import InMemorySaver

# 1. 初始化 Tavily 搜索工具
tavily = TavilySearch(max_results=3)

# 2. 编写系统提示词
research_instructions = """
您是一位资深的研究人员。您的工作是进行深入的研究，然后撰写一份精美的报告。
您可以通过互联网搜索引擎作为主要的信息收集工具。
## 可用工具
### `互联网搜索`
使用此功能针对给定的查询进行互联网搜索。您可以指定要返回的最大结果数量、主题以及是否包含原始内容。
### `写入本地文件`
使用此功能将研究报告保存到本地文件。当您完成研究并生成报告后，请使用此工具将完整的报告内容保存到文
件中。
- 文件路径建议使用 .md 格式（Markdown），例如 "research_report.md" 或 "./reports/报告名
称.md"
- 请确保报告内容完整、结构清晰，包含所有章节和引用来源
## 工作流程
在进行研究时：
1. 首先将研究任务分解为清晰的步骤
2. 使用互联网搜索来收集全面的信息
3. 将信息整合成一份结构清晰的报告
4. **重要**：完成报告后，务必使用 `写入本地文件` 工具将完整报告保存到本地文件
5. 务必引用你的资料来源
**注意**：请确保在完成研究后，将完整的报告内容保存到文件中，这样用户可以方便地查看和保存报告。
"""

# 2. 创建 DeepAgents 智能体
agent = create_deep_agent(
    name="DeepAgents_Agent",       # 智能体名称
    tools=[tavily],                # 可调用工具Tool
    model=model,                   # 模型Model
    system_prompt=research_instructions,  # 系统提示词
    checkpointer=InMemorySaver(),  # 检查点Checkpointer，内存检查点
)

# 3. 配置线程 ID
config = {"configurable": {"thread_id": "1"}}

result = agent.invoke({"messages": [{"role": "user", "content": "帮我查询一下有关deepagents框架的最新动态"}]}, config=config)

print(result["messages"][-1].content)

# 安装美化代码库Rich
!pip install rich

# 导入Rich库，用于美化代码
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
RICH_AVAILABLE = True
console = Console()

def print_agent_tools(agent):
    """
    打印 Agent 中加载的所有工具
    包括用户自定义工具、文件系统工具、系统工具等
    """
    # 获取 agent 的 nodes (LangGraph 的节点)
    if hasattr(agent, 'nodes') and 'tools' in agent.nodes:
        tools_node = agent.nodes['tools']

        # tools_node 是 PregelNode，真正的 ToolNode 在 bound 属性中
        if hasattr(tools_node, 'bound'):
            tool_node = tools_node.bound

            # 从 ToolNode 获取工具
            if hasattr(tool_node, 'tools_by_name'):
                tools = tool_node.tools_by_name

                # 分类工具
                user_tools = []
                filesystem_tools = []
                system_tools = []

                for tool_name, tool in tools.items():
                    tool_info = {
                        'name': tool_name,
                        'description': getattr(tool, 'description', '无描述')
                    }

                    # 分类
                    if tool_name in ['ls', 'read_file', 'write_file', 'edit_file', 'glob', 'grep', 'execute']:
                        filesystem_tools.append(tool_info)
                    elif tool_name in ['write_todos', 'task']:
                        system_tools.append(tool_info)
                    else:
                        user_tools.append(tool_info)

                # 打印加载工具的输出
                _print_tools_rich(user_tools, filesystem_tools, system_tools)

            else:
                print("无法获取工具列表 (tools_by_name 不存在)")
        else:
            print("无法获取工具列表 (bound 属性不存在)")
    else:
        print("无法获取工具列表 (nodes 结构不符合预期)")

def _print_tools_rich(user_tools, filesystem_tools, system_tools):
    """使用 Rich 库美化打印工具列表"""
    console.print()

    # 创建表格
    table = Table(title="Agent 加载的工具列表", show_header=True, header_style="bold magenta")
    table.add_column("类别", style="cyan", width=20)
    table.add_column("工具名称", style="green", width=20)
    table.add_column("描述", style="white", width=60)

    # 添加用户工具
    for i, tool in enumerate(user_tools):
        category = "用户工具" if i == 0 else ""
        desc = tool['description'][:80] + "..." if len(tool['description']) > 80 else tool['description']
        table.add_row(category, tool['name'], desc)

    # 添加文件系统工具
    for i, tool in enumerate(filesystem_tools):
        category = "文件系统工具" if i == 0 else ""
        desc = tool['description'][:80] + "..." if len(tool['description']) > 80 else tool['description']
        table.add_row(category, tool['name'], desc)

    # 添加系统工具
    for i, tool in enumerate(system_tools):
        category = "系统工具" if i == 0 else ""
        desc = tool['description'][:80] + "..." if len(tool['description']) > 80 else tool['description']
        table.add_row(category, tool['name'], desc)

    console.print(table)

    # 打印统计
    total = len(user_tools) + len(filesystem_tools) + len(system_tools)
    console.print(Panel(
        f"[bold green]共计 {total} 个工具[/bold green]\n\n"
        f"• 用户工具: {len(user_tools)} 个\n"
        f"• 文件系统工具: {len(filesystem_tools)} 个\n"
        f"• 系统工具: {len(system_tools)} 个",
        title="统计信息",
        border_style="green"
    ))
    console.print()

print_agent_tools(agent)

import json
from rich.json import JSON

def debug_agent(query: str, save_to_file: str = None):
    """
    运行智能体并打印中间过程（使用 Rich 美化输出）
    参数:
    query: 用户查询
    save_to_file: 保存最终输出到文件(可选)
    返回:
    str: 最终的研究报告
    """
    console.print(Panel.fit(
        f"[bold cyan]查询:[/bold cyan] {query}",
        border_style="cyan"
    ))
    step_num = 0
    final_response = None

    config = {"configurable": {"thread_id": "2"}}

    # 实时流式输出
    for event in agent.stream(
        {"messages": [{"role": "user", "content": query}]},
        stream_mode="values",
        config=config
    ):
        step_num += 1

        console.print(f"\n[bold yellow]{'─' * 80}[/bold yellow]")
        console.print(f"[bold yellow]步骤 {step_num}[/bold yellow]")
        console.print(f"[bold yellow]{'─' * 80}[/bold yellow]")

        if "messages" in event:
            messages = event["messages"]
            if messages:
                msg = messages[-1]
                # 保存最终响应
                if hasattr(msg, 'content') and msg.content and not hasattr(msg,'tool_calls'):
                    final_response = msg.content
                # AI 思考
                if hasattr(msg, 'content') and msg.content:
                    # 如果内容太长,只显示前300字符作为预览
                    content = msg.content
                    if len(content) > 300 and not (hasattr(msg, 'tool_calls') and msg.tool_calls):
                        preview = content[:300] + "..."
                        console.print(Panel(
                            f"{preview}\n\n[dim](内容较长,完整内容将在最后显示)[/dim]",
                        title="[bold green]AI 思考[/bold green]",
                        border_style="green"
                        ))
                    else:
                        console.print(Panel(
                            content,
                            title="[bold green]AI 思考[/bold green]",
                            border_style="green"
                        ))
                # 工具调用
                if hasattr(msg, 'tool_calls') and msg.tool_calls:
                    for tool_call in msg.tool_calls:
                        tool_info = {
                            "工具名称": tool_call.get('name', 'unknown'),
                            "参数": tool_call.get('args', {})
                        }
                        console.print(Panel(
                            JSON(json.dumps(tool_info, ensure_ascii=False)),
                            title="[bold blue]工具调用[/bold blue]",
                            border_style="blue"
                        ))
                # 工具响应
                if hasattr(msg, 'name') and msg.name:
                    response = str(msg.content)[:500]
                    if len(str(msg.content)) > 500:
                        response += f"\n... (共 {len(str(msg.content))} 字符)"

                    console.print(Panel(
                        response,
                        title=f"[bold magenta]工具响应: {msg.name}[/bold magenta]",
                        border_style="magenta"
                    ))
    console.print("\n[bold green]任务完成![/bold green]\n")
    return final_response

print("调试函数已创建")

# 示例：使用调试函数运行研究任务
query = "详细调研 LangChain DeepAgents 框架的核心特性，并写一份结构化的总结报告。"

# 使用调试函数）
result = debug_agent(query)

import asyncio
import os
from dotenv import load_dotenv
from rich.console import Console
from rich.panel import Panel
from rich.tree import Tree
from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI
from langchain_community.tools import TavilySearchResults

from langchain_core.messages import ToolMessage, BaseMessage

# 加载环境变量
load_dotenv(override=True)

# 配置 Rich Console
console = Console()

async def run_auto_subagent_demo():
    """
    演示：不显式传入 subagents 参数，自动触发默认的 SubAgentMiddleware
    """
    console.print(Panel.fit("[bold magenta]DeepAgents 自动 SubAgent 中间件演示[/bold magenta]", border_style="magenta"))
    console.print("[dim]本演示验证：即使不传入 subagents 参数，Agent 默认也会启用 'general-purpose' 子 Agent。[/dim]")

    # 1. 初始化模型
    # 使用 GPT-4o 以确保对复杂指令的理解和工具调用的准确性
    llm = ChatOpenAI(model="gpt-4o", temperature=0)

    # 2. 定义工具
    # 给 Agent 一些工具，以便子 Agent 也有工具可用
    tools = [TavilySearchResults(max_results=2)]

    # 3. 创建 Agent (不传入 subagents)
    # create_deep_agent 默认会加载 SubAgentMiddleware(general_purpose_agent=True)
    # 这意味着 Agent 会自动获得一个名为 'task' 的工具，可以调用 'general-purpose' 子 Agent
    console.print("[bold cyan]正在创建 Agent (subagents=None)...[/bold cyan]")

    agent = create_deep_agent(
        model=llm,
        tools=tools,
        # subagents=[], # 故意不传或传空
        system_prompt="""你是一个能够高效处理并发任务的智能助手。
                        对于包含多个独立部分的复杂任务，你必须使用 'task' 工具来创建 'general-purpose' 子 Agent 进行处理。
                        不要自己在主线程中串行执行所有操作。利用子 Agent 来隔离上下文并提高效率。"""
    )

    # 4. 定义一个适合并行/隔离的任务
    task = """请同时调研以下两个完全不同的主题，并分别给出简短总结：
            1. Python 语言的历史起源。
            2. Rust 语言的内存安全机制。
            请务必使用子 Agent 分别处理这两个任务。"""

    console.print(f"\n[bold green]用户任务:[/bold green] {task}\n")

    # 5. 运行并可视化
    step = 0

    # 检查 API Key
    if not os.getenv("OPENAI_API_KEY"):
        console.print("[bold red]❌ 错误: 未找到 OPENAI_API_KEY，请检查 .env 文件[/bold red]")
        return

    console.print("[dim]开始流式输出...[/dim]")
    try:
        async for event in agent.astream({"messages": [("user", task)]}):
            step += 1
            # 遍历所有节点输出 (e.g., 'agent', 'tools')
            for node_name, node_data in event.items():
                if node_data is None:
                    continue

                if "messages" in node_data:
                    msgs = node_data["messages"]
                    # 确保是列表
                    if not isinstance(msgs, list):
                        msgs = [msgs]

                    for msg in msgs:
                        # 0. 过滤非消息对象
                        if not isinstance(msg, BaseMessage):
                            continue

                        # 1. 检测工具调用 (期望看到 'task' 工具)
                        if hasattr(msg, "tool_calls") and msg.tool_calls:
                            tree = Tree(f"[bold yellow]Step {step}: 决策与调用 (Node: {node_name})[/bold yellow]")
                            for tc in msg.tool_calls:
                                tool_name = tc['name']
                                tool_args = tc['args']

                                if tool_name == "task":
                                    # 验证成功！
                                    branch = tree.add(f"[bold red]🚀 触发 'task' 工具 (Sub-Agent)[/bold red]")
                                    branch.add(f"[cyan]子 Agent 类型:[/cyan] {tool_args.get('subagent_type')}")
                                    branch.add(f"[cyan]任务指令:[/cyan] {tool_args.get('description')}")
                                else:
                                    tree.add(f"[blue]普通工具调用:[/blue] {tool_name}")

                            console.print(tree)

                        # 2. 检测工具输出 (Sub-Agent 的返回结果)
                        elif isinstance(msg, ToolMessage):
                            if msg.name == "task":
                                # Sub-Agent 完成任务返回
                                panel = Panel(
                                    msg.content,
                                    title=f"[bold magenta]Sub-Agent 完成任务 (Node: {node_name})[/bold magenta]",
                                    border_style="magenta"
                                )
                                console.print(panel)
                            else:
                                console.print(f"[dim]Tool Output ({msg.name}): {msg.content[:100]}...[/dim]")

                        # 3. 检测 AI 最终回复
                        elif msg.content and not msg.tool_calls:
                            title = f"[bold green]Agent 回复 (Node: {node_name})[/bold green]"
                            console.print(Panel(msg.content, title=title, border_style="green"))

    except Exception as e:
        console.print(f"[bold red]❌ 运行时错误: {e}[/bold red]")

    console.print("\n[bold magenta]演示结束[/bold magenta]")

if __name__ == "__main__":
    await run_auto_subagent_demo()

# 安装 MCP 适配器（关键依赖）\MCP 服务器开发库（如需自定义工具）
#!pip install langchain-mcp-adapters mcp

import asyncio
import os
from dotenv import load_dotenv
from rich.console import Console
from rich.panel import Panel
from rich.tree import Tree
from deepagents import create_deep_agent
from langchain_deepseek import ChatDeepSeek
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_community.tools import TavilySearchResults

from langchain_core.messages import ToolMessage, BaseMessage

# 加载环境变量
load_dotenv(override=True)

# 1. 配置 Rich Console
console = Console()

# 2. 配置 Context7 MCP (连接官方文档)
async def setup_mcp_tools():
    console.print("[dim]正在连接 Context7 MCP 服务器...[/dim]")
    # 检查 node 环境
    try:
        client = MultiServerMCPClient({
            "context7": {
                "transport": "stdio",
                "command": "npx",
                "args": ["-y", "@upstash/context7-mcp@latest"],
            }
        })
        # 获取工具
        tools = await client.get_tools()
        console.print(f"[green]成功加载 {len(tools)} 个 MCP 工具[/green]")
        return client, tools
    except Exception as e:
        console.print(f"[red]连接 MCP 失败: {e}[/red]")
        console.print("[yellow]将使用模拟工具继续...[/yellow]")
        return None, []

# 3. 定义子 Agent 配置
def get_subagents_config(mcp_tools):
    # 子 Agent 1: 官方文档专家
    doc_tools = mcp_tools if mcp_tools else [TavilySearchResults(max_results=3)]

    docs_researcher = {
        "name": "DocsResearcher",
        "description": "负责查阅官方文档和技术规范的专家 Agent。",
        "system_prompt": "你是一名专门查阅官方文档的技术专家。请使用工具获取准确的技术细节。不要猜测。",
        "tools": doc_tools,
        "model": "deepseek-chat"
    }

    # 子 Agent 2: 社区生态专家
    community_researcher = {
        "name": "CommunityResearcher",
        "description": "负责搜索社区博客、教程和最佳实践的专家 Agent。",
        "system_prompt": "你是一名关注社区动态的开发者。请搜索博客、论坛和 GitHub 讨论。",
        "tools": [TavilySearchResults(max_results=3)],
        "model": "deepseek-chat"
    }

    return [docs_researcher, community_researcher]

# 4. 主运行逻辑
async def run_parallel_demo():
    console.print(Panel.fit("[bold blue]DeepAgents 并行子 Agent 演示[/bold blue]", border_style="blue"))

    # 初始化 MCP
    mcp_client, mcp_tools = await setup_mcp_tools()

    # 获取子 Agent 配置
    subagents = get_subagents_config(mcp_tools)

    # 创建主 Agent
    llm = ChatDeepSeek(model="deepseek-chat", temperature=0)

    agent = create_deep_agent(
        model=llm,
        tools=[],
        subagents=subagents,        # 传入子 Agent 配置
        system_prompt="""你是一名技术总监。你的任务是协调 DocsResearcher 和 CommunityResearcher 完成调研任务。
                        请根据用户需求，将任务拆解并分发给这两个子 Agent。
                        如果任务允许，请务必并行调用它们以提高效率。
                        最后汇总它们的报告。"""
    )

    task = "请详细调研 'LangChain DeepAgents' 框架。我需要官方的技术架构说明（来自文档）以及社区的最佳实践案例。请对比两者。"

    console.print(f"\n[bold green]任务指令:[/bold green] {task}\n")

    # 运行并可视化
    step = 0
    try:
        async for event in agent.astream({"messages": [("user", task)]}):
            step += 1
            # 遍历所有节点输出 (e.g., 'agent', 'tools')
            for node_name, node_data in event.items():
                if node_data is None:
                    continue

                if "messages" in node_data:
                    msgs = node_data["messages"]
                    # 确保是列表
                    if not isinstance(msgs, list):
                        msgs = [msgs]

                    for msg in msgs:
                        # 0. 过滤非消息对象
                        if not isinstance(msg, BaseMessage):
                            continue

                        # 1. 检测工具调用 (期望看到 'task' 工具)
                        if hasattr(msg, "tool_calls") and msg.tool_calls:
                            tree = Tree(f"[bold yellow]Step {step}: 决策与调用 (Node: {node_name})[/bold yellow]")
                            for tc in msg.tool_calls:
                                tool_name = tc['name']
                                tool_args = tc['args']

                                if tool_name == "task":
                                    # 验证成功！
                                    branch = tree.add(f"[bold red]🚀 触发 'task' 工具 (Sub-Agent)[/bold red]")
                                    branch.add(f"[cyan]子 Agent 类型:[/cyan] {tool_args.get('subagent_type')}")
                                    branch.add(f"[cyan]任务指令:[/cyan] {tool_args.get('description')}")
                                else:
                                    tree.add(f"[blue]普通工具调用:[/blue] {tool_name}")

                            console.print(tree)

                        # 2. 检测工具输出 (Sub-Agent 的返回结果)
                        elif isinstance(msg, ToolMessage):
                            if msg.name == "task":
                                # Sub-Agent 完成任务返回
                                panel = Panel(
                                    msg.content,
                                    title=f"[bold magenta]Sub-Agent 完成任务 (Node: {node_name})[/bold magenta]",
                                    border_style="magenta"
                                )
                                console.print(panel)
                            else:
                                console.print(f"[dim]Tool Output ({msg.name}): {msg.content[:100]}...[/dim]")

                        # 3. 检测 AI 最终回复
                        elif msg.content and not msg.tool_calls:
                            title = f"[bold green]Agent 回复 (Node: {node_name})[/bold green]"
                            console.print(Panel(msg.content, title=title, border_style="green"))

    except Exception as e:
        console.print(f"[bold red]❌ 运行时错误: {e}[/bold red]")


    console.print("\n[bold blue]演示结束[/bold blue]")

# 运行演示
if __name__ == "__main__":
    await run_parallel_demo()

# --- FilesystemMiddleware 工具测试 ---
print("\n" + "="*50)
print("🚀 开始测试 FilesystemMiddleware 工具集")
print("="*50 + "\n")

from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

# 1. 初始化环境
load_dotenv(override=True)
model = ChatOpenAI(model="gpt-4o", temperature=0)

# 使用本地文件系统后端，根目录设为 ./workspace
# 这样我们可以看到真实文件的创建
# 注意：设置 virtual_mode=True 以支持绝对路径 (如 /hello_world.py) 映射到 ./workspace
backend = FilesystemBackend(root_dir="./workspace", virtual_mode=True)

# 创建 Agent
# 注意：create_deep_agent 默认会自动包含 FilesystemMiddleware
agent = create_deep_agent(
    model=model,
    backend=backend,
    system_prompt="你是一个文件系统操作助手。请根据用户指令使用相应的工具。"
)

# 辅助函数：运行并打印结果
def run_test(task_name, instruction):
    print(f"\n🔹 [测试: {task_name}]")
    print(f"指令: {instruction}")
    try:
        # 使用 invoke 而不是 stream 以简化输出
        result = agent.invoke({"messages": [("user", instruction)]})
        last_msg = result["messages"][-1]
        print(f"🤖 Agent 回复: {last_msg.content}")

        # 打印工具调用详情 (如果有)
        for msg in result["messages"]:
            if hasattr(msg, 'tool_calls') and msg.tool_calls:
                for tool in msg.tool_calls:
                    print(f"🛠️  调用工具: {tool['name']} args={tool['args']}")
    except Exception as e:
        print(f"❌ 发生错误: {e}")

# 2. 依次测试各个工具

# 案例 1: write_file
# 创建一个 Python 脚本文件
run_test("write_file",
         "请在当前目录下创建一个名为 'hello_world.py' 的文件，内容是：\nprint('Hello from DeepAgents!')")

# 案例 2: ls
# 查看目录内容，确认文件创建成功
run_test("ls", "请列出当前目录下的所有文件，确认 hello_world.py 是否存在。")

# 案例 3: read_file
# 读取刚才创建的文件内容
run_test("read_file", "请读取 'hello_world.py' 的内容并展示给我。")

# 案例 4: edit_file
# 修改文件内容
run_test("edit_file",
         "请修改 'hello_world.py' 文件，将 print 内容改为 'Hello from Modified File!'。")

# 案例 5: grep
# 搜索文件中的特定字符串
run_test("grep", "请在当前目录下搜索包含 'Modified' 字符串的文件。")

# 案例 6: glob
# 使用通配符查找文件
run_test("glob", "请找出当前目录下所有的 .py 文件。")

# 案例 7: execute
# 尝试执行脚本
# 注意：execute 工具通常需要 SandboxBackend 支持。
# 如果当前使用的 FilesystemBackend 不支持执行，Agent 会收到错误提示或无法调用。
print("\n🔹 [测试: execute]")
print("指令: 尝试运行 hello_world.py 脚本")
try:
    # 我们尝试强行要求 Agent 运行，看它如何反应
    # 如果没有 execute 工具，Agent 可能会说无法执行
    response = agent.invoke({"messages": [("user", "请使用 execute 工具运行 python hello_world.py")]})
    print(f"🤖 Agent 回复: {response['messages'][-1].content}")
except Exception as e:
    print(f"⚠️ 测试说明: execute 工具可能不可用 (取决于 Backend 支持): {e}")

print("\n" + "="*50)
print("✅ 测试结束")

from deepagents import create_deep_agent
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

# 加载环境变量
load_dotenv(override=True)

# 定义 LLM
model = ChatOpenAI(model="gpt-4o", temperature=0)

# 默认情况下，DeepAgent 会自动加载 FilesystemMiddleware 并使用内存后端
agent = create_deep_agent(
    model=model,
    system_prompt="你是一个数据处理助手。"
)

# Agent 可以自由创建文件、读取文件，但这些文件只存在于内存中
# 任务结束后，这些文件会自动消失

agent.invoke({"messages": [("user", "创建一个名为 test.txt 的文件并写入 'Hello, World!'")]})

result = agent.invoke({"messages": [("user", "读取 test.txt 文件的内容")]})
print(result["messages"][-1].content)

from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o", temperature=0)

# 将 Agent 的根目录映射到本地的 "./workspace" 文件夹，virtual_mode=True 表示启用虚拟文件系统，作用是将 Agent 执行的所有文件操作都映射到本地的 "./workspace" 文件夹，而不是直接操作本地文件系统。
backend = FilesystemBackend(root_dir="./workspace", virtual_mode=True)

agent = create_deep_agent(
    model=model,
    backend=backend  # 传入后端，中间件会自动使用它
)

# 此时 Agent 执行 write_file("/readme.md", ...) 会在本地 ./workspace/readme.md 创建文件
agent.invoke({"messages": [("user", "创建一个名为 readme.md 的文件并写入 'Hello, World!'")]})

result = agent.invoke({"messages": [("user", "读取 readme.md 文件的内容")]})
print(result["messages"][-1].content)

import asyncio
import os
import shutil
from pathlib import Path
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, ToolMessage
from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend

load_dotenv(override=True)

# 定义工作目录
WORK_DIR = Path("workspace/pagination_demo").resolve()
LARGE_FILE_NAME = "server_logs.txt"
LARGE_FILE_PATH = WORK_DIR / LARGE_FILE_NAME
TARGET_SECRET = "CRITICAL_ERROR_CODE_998877"

async def run_pagination_demo():
    print("\n" + "="*80)
    print("DeepAgents FilesystemMiddleware 分页读取演示")
    print("="*80)

    # 2. 初始化 FilesystemBackend
    # 将 backend 指向我们的测试目录
    # virtual_mode=True 确保 Agent 只能访问该目录下的文件，不能访问宿主机其他目录
    backend = FilesystemBackend(root_dir=WORK_DIR, virtual_mode=True)

    # 3. 创建 Agent
    # 我们明确指示 Agent 使用分页读取，每次读取 300 行，Agent默认每次读取500行(DEFAULT_READ_LIMIT)
    system_prompt = """
    你是一个专业的系统管理员。
    你的任务是从日志文件中查找特定的错误代码。
    注意：日志文件可能非常大，为了避免上下文溢出，你必须使用 `read_file` 工具的分页功能。
    每次读取请限制在 300 行以内 (limit=300)，并使用 offset 参数向后滚动。
    直到找到目标信息为止。
    """

    llm = ChatOpenAI(model="gpt-4o", temperature=0)

    agent = create_deep_agent(
        model=llm,
        backend=backend,
        system_prompt=system_prompt
    )

    task = f"请在 '/{LARGE_FILE_NAME}' 中查找包含 '{TARGET_SECRET}' 的行，并告诉我它的具体内容。"

    print(f"\n任务: {task}")
    print("-" * 60)

    # 4. 执行任务并观察分页行为
    step = 0
    print("\n开始流式输出...")
    try:
        async for event in agent.astream({"messages": [("user", task)]}):
            for node_name, node_data in event.items():
                # debug: print(f"DEBUG: Node: {node_name}")
                if not node_data: continue

                # 处理 Overwrite 对象
                if hasattr(node_data, "value"):
                    node_data = node_data.value

                if not isinstance(node_data, dict):
                    continue

                if "messages" in node_data:
                    msgs = node_data["messages"]
                    if hasattr(msgs, "value"):
                        msgs = msgs.value

                    if not isinstance(msgs, list): msgs = [msgs]

                    for msg in msgs:
                        # 1. 打印 Agent 的思考 (AIMessage with tool_calls)
                        if hasattr(msg, "tool_calls") and msg.tool_calls:
                            step += 1
                            print(f"\n[Step {step}] Agent 决定调用工具 (Node: {node_name}):")
                            for tc in msg.tool_calls:
                                name = tc['name']
                                args = tc['args']
                                print(f"  >>> 工具: {name}")

                                if name == "read_file":
                                    offset = args.get('offset', 0)
                                    limit = args.get('limit', 'Default')
                                    path_val = args.get('path') or args.get('file_path')
                                    print(f"  >>> 参数: path='{path_val}', offset={offset}, limit={limit}")
                                    print(f"      (说明: 正在读取从第 {offset} 行开始的 {limit} 行数据)")
                                else:
                                    print(f"  >>> 参数: {args}")

                        # 2. 打印工具的输出 (ToolMessage)
                        elif isinstance(msg, ToolMessage):
                            content = msg.content
                            line_count = len(content.splitlines())
                            # 检查是否包含目标 Secret
                            found_secret = TARGET_SECRET in content

                            preview = content[:100].replace('\n', ' ') + "..."
                            print(f"\n[Tool Output] (Node: {node_name}) 读取了 {line_count} 行数据")
                            print(f"  内容预览: {preview}")
                            if found_secret:
                                print(f"  ✨ 成功: 在此分块中发现了目标 Secret: {TARGET_SECRET}")

                        # 3. 打印 Agent 的最终回复 (AIMessage without tool_calls)
                        elif isinstance(msg, BaseMessage) and msg.type == "ai" and msg.content:
                            print(f"\n[Agent 最终回复] (Node: {node_name}):")
                            print("-" * 40)
                            print(msg.content)
                            print("-" * 40)


    except Exception as e:
        print(f"❌ 运行出错: {e}")
        import traceback
        traceback.print_exc()

if __name__ == "__main__":
    await run_pagination_demo()

# 定义 E2B 沙箱后端类
# 该类继承自 BaseSandbox，实现了在 E2B 沙箱中执行命令的功能

import base64
from typing import Any, Optional

# 导入 DeepAgents 后端协议，定义了沙箱后端的接口
from deepagents.backends.protocol import (
    ExecuteResponse,
    FileDownloadResponse,
    FileUploadResponse,
    SandboxBackendProtocol,
)
# 导入基础沙箱类，用于实现沙箱后端的基本功能
from deepagents.backends.sandbox import BaseSandbox

try:
    from e2b import Sandbox
except ImportError:
    Sandbox = None

class E2BBackend(BaseSandbox):
    """E2B 沙箱后端实现，用于 DeepAgents。

    该后端使用 E2B（https://e2b.dev）提供安全、隔离的执行环境。
    """

    def __init__(
        self,
        template: str = "base",
        api_key: Optional[str] = None,
        timeout: Optional[int] = None,
        metadata: Optional[dict[str, str]] = None,
    ) -> None:
        """初始化 E2B 沙箱。

        参数:
            template: E2B 沙箱模板 ID（默认："base"）
            api_key: E2B API 密钥（可选，默认使用 E2B_API_KEY 环境变量）
            timeout: 沙箱超时时间（秒）
            metadata: 自定义沙箱元数据
        """
        if Sandbox is None:
            raise ImportError(
                "e2b package is not installed. "
                "Please install it with `pip install e2b`."
            )

        self.sandbox = Sandbox.create(
            template=template,
            api_key=api_key,
            timeout=timeout,
            metadata=metadata,
        )

    @property
    def id(self) -> str:
        """Unique identifier for the sandbox backend."""
        return self.sandbox.sandbox_id

    def execute(self, command: str) -> ExecuteResponse:
        """Execute a command in the sandbox."""
        try:
            # E2B commands.run returns CommandResult with stdout, stderr, exit_code
            result = self.sandbox.commands.run(command)

            # 返回执行结果，包含 stdout 是标准输出，stderr 是标准错误输出，exit_code 是退出码
            return ExecuteResponse(
                output=result.stdout + result.stderr,
                exit_code=result.exit_code,
                truncated=False,
            )
        except Exception as e:
            return ExecuteResponse(
                output=f"Error executing command: {str(e)}",
                exit_code=1,
                truncated=False,
            )

    def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]:
        """Upload multiple files to the sandbox."""
        responses = []
        for path, content in files:
            try:
                # Ensure directory exists before writing
                # We can use execute to mkdir -p
                parent_dir = path.rsplit("/", 1)[0]
                if parent_dir:
                    self.sandbox.commands.run(f"mkdir -p {parent_dir}")

                # Write file
                self.sandbox.files.write(path, content)
                responses.append(FileUploadResponse(path=path, error=None))
            except Exception as e:
                error_msg = str(e).lower()
                error = "invalid_path"
                if "permission" in error_msg:
                    error = "permission_denied"

                responses.append(FileUploadResponse(path=path, error=error))
        return responses

    def download_files(self, paths: list[str]) -> list[FileDownloadResponse]:
        """Download multiple files from the sandbox."""
        responses = []
        for path in paths:
            try:
                content = self.sandbox.files.read(path)
                # Ensure content is bytes
                if isinstance(content, str):
                    content = content.encode("utf-8")

                responses.append(FileDownloadResponse(path=path, content=content, error=None))
            except Exception as e:
                error_msg = str(e).lower()
                error = "invalid_path"
                if "not found" in error_msg:
                    error = "file_not_found"
                elif "directory" in error_msg:
                    error = "is_directory"
                elif "permission" in error_msg:
                    error = "permission_denied"

                responses.append(FileDownloadResponse(path=path, content=None, error=error))
        return responses

    def close(self):
        """Close the sandbox session."""
        self.sandbox.kill()

import asyncio
import os
import sys
from dotenv import load_dotenv
from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_core.messages import BaseMessage, ToolMessage

# 假设 e2b_backend.py 在同一目录下
try:
    from e2b_backend import E2BBackend
except ImportError:
    # 尝试从当前路径导入
    sys.path.append(os.path.dirname(os.path.abspath(__file__)))
    from e2b_backend import E2BBackend

# 加载环境变量
load_dotenv(override=True)

async def setup_mcp_tools():
    """连接 Context7 MCP 服务器并获取工具"""
    print("正在连接 Context7 MCP 服务器...")
    try:
        # 使用官方 Context7 MCP 配置
        client = MultiServerMCPClient({
            "context7": {
                "transport": "stdio",
                "command": "npx",
                "args": ["-y", "@upstash/context7-mcp@latest"],
            }
        })
        # 获取工具
        tools = await client.get_tools()
        print(f"成功加载 {len(tools)} 个 MCP 工具")
        return client, tools
    except Exception as e:
        print(f"❌ 连接 MCP 失败: {e}")
        return None, []

async def run_e2b_demo():
    print("\n" + "="*80)
    print("DeepAgents E2BBackend (Sandbox) 演示 ")
    print("="*80)

    # 1. 检查 API Key
    if not os.getenv("E2B_API_KEY"):
        print("❌ 错误: 未找到 E2B_API_KEY 环境变量。")
        print("请在 .env 文件中设置 E2B_API_KEY，或直接导出该变量。")
        return

    # 2. 初始化 MCP (用于查询文档等辅助任务)
    mcp_client, mcp_tools = await setup_mcp_tools()

    # 3. 初始化 E2B Backend
    print("正在初始化 E2B 沙箱 (Template: base)...")
    try:
        backend = E2BBackend(template="base")
        print(f"沙箱已启动 (ID: {backend.id})")
    except Exception as e:
        print(f"❌ 沙箱启动失败: {e}")
        return

    try:
        # 4. 创建 DeepAgent
        llm = ChatOpenAI(model="gpt-4o", temperature=0)

        agent = create_deep_agent(
            model=llm,
            tools=mcp_tools,     # 赋予 MCP 工具能力
            backend=backend,     # 赋予 E2B 沙箱能力
            system_prompt="""你是一个拥有云端沙箱环境的高级技术助手。
            你的任务是演示如何在沙箱中进行操作。
            请执行以下步骤：
            1. 使用 'execute_command' 运行 'uname -a' 和 'python --version' 来展示环境信息。
            2. 创建一个 Python 脚本 '/home/user/hello.py'，内容是打印 'Hello from E2B Sandbox!'。
            3. 运行这个 Python 脚本并显示输出。
            """
        )

        print(f"\n任务开始")

        # 5. 执行任务
        task = "请开始演示沙箱操作流程。"

        step = 0
        async for event in agent.astream({"messages": [("user", task)]}):
            step += 1
            for node_name, node_data in event.items():
                if node_data is None: continue

                if "messages" in node_data:
                    msgs = node_data["messages"]
                    if not isinstance(msgs, list): msgs = [msgs]

                    for msg in msgs:
                        if isinstance(msg, BaseMessage) and msg.content:
                            # 过滤掉空的工具调用消息
                            is_tool_call_msg = getattr(msg, "tool_calls", None)
                            if not is_tool_call_msg:
                                print(f"\n[Agent ({node_name})]")
                                print("-" * 40)
                                print(msg.content)
                                print("-" * 40)

                        if hasattr(msg, "tool_calls") and msg.tool_calls:
                            print(f"\n[Step {step}: 工具调用]")
                            for tc in msg.tool_calls:
                                args_str = str(tc['args'])
                                if len(args_str) > 500:
                                    args_str = args_str[:500] + "..."
                                print(f"  • {tc['name']}: {args_str}")

                        if isinstance(msg, ToolMessage):
                            content_preview = str(msg.content)
                            if len(content_preview) > 200:
                                content_preview = content_preview[:200] + "..."
                            print(f"[Tool Output ({msg.name})]: {content_preview}")

    except Exception as e:
        print(f"❌ 运行时错误: {e}")
        import traceback
        traceback.print_exc()
    finally:
        # 6. 清理资源
        print("\n正在关闭沙箱...")
        backend.close()
        print("沙箱已关闭")
        print("演示结束")

if __name__ == "__main__":
    try:
        # asyncio.run(run_e2b_demo())
        await run_e2b_demo()
    except RuntimeError as e:
        if "asyncio.run() cannot be called from a running event loop" in str(e):
            print("请在 Jupyter 中使用 await run_e2b_demo()")
        else:
            raise e

# 需要先安装一下docker依赖
#!pip install docker

#这个 Backend 会在本地启动一个 Docker 容器，并将会话隔离在容器内部。

# - 核心功能 ：
#   - 自动生命周期管理 ：初始化时启动容器，结束时自动销毁 ( auto_remove=True )。
#   - 高效文件传输 ：使用 tar 流在宿主机和容器之间传输文件，支持批量操作。
#   - 资源限制 ：支持设置 CPU ( cpu_quota ) 和内存 ( memory_limit ) 限制，防止 Agent 耗尽本机资源。
#   - 网络控制 ：可选禁用网络 ( network_disabled=True ) 以增强安全性。

import io
import tarfile
import time
import uuid
from typing import Optional

# 加载 DeepAgents 后端协议
from deepagents.backends.protocol import (
    ExecuteResponse,
    FileDownloadResponse,
    FileUploadResponse,
    SandboxBackendProtocol,
)

# 加载 DeepAgents 基础沙箱类
from deepagents.backends.sandbox import BaseSandbox

try:
    import docker
    from docker.errors import NotFound, APIError
except ImportError:
    docker = None

class DockerBackend(BaseSandbox):
    """Docker 沙箱后端实现，用于 DeepAgents。

    该后端使用本地 Docker 守护进程提供隔离的执行环境。
    需要安装 `docker` Python 包，并确保 Docker 守护进程正在运行。
    """

    def __init__(
        self,
        image: str = "python:3.11-slim",
        auto_remove: bool = True,
        cpu_quota: int = 50000,  # 50% CPU
        memory_limit: str = "512m",
        network_disabled: bool = False,
        working_dir: str = "/workspace",
        volumes: dict[str, dict[str, str]] | None = None,
    ) -> None:
        """初始化 Docker 沙箱。

        参数:
            image: 使用的 Docker 镜像（默认："python:3.11-slim"）
            auto_remove: 是否在关闭时移除容器（默认：True）
            cpu_quota: CPU 配额，单位为微秒（默认：50000）
            memory_limit: 内存限制（默认："512m"）
            network_disabled: 是否禁用网络访问（默认：False）
            working_dir: 容器内的工作目录（默认："/workspace"）
            volumes: Docker 卷配置，例如 {'/宿主机路径': {'bind': '/容器路径', 'mode': 'rw'}}
        """
        if docker is None:
            raise ImportError(
                "docker package is not installed. "
                "Please install it with `pip install docker`."
            )
        # 初始化 Docker 客户端，from_env() 会自动从环境变量中读取 Docker 配置
        self.client = docker.from_env()
        self.image = image
        self.auto_remove = auto_remove
        self.working_dir = working_dir
        self.volumes = volumes or {}
        self._container = None

        # Start container
        try:
            # Ensure image exists
            try:
                self.client.images.get(image)
            except NotFound:
                print(f"Pulling image {image}...")
                self.client.images.pull(image)

            self._container = self.client.containers.run(
                image,
                command="tail -f /dev/null",  # Keep container running
                detach=True,
                tty=True,
                cpu_quota=cpu_quota,
                mem_limit=memory_limit,
                network_disabled=network_disabled,
                working_dir=working_dir,
                volumes=self.volumes,
            )

            # Ensure working directory exists
            self.execute(f"mkdir -p {working_dir}")

        except Exception as e:
            raise RuntimeError(f"Failed to start Docker container: {e}")

    @property
    def id(self) -> str:
        """Unique identifier for the sandbox backend."""
        return self._container.id if self._container else "unknown"

    def execute(self, command: str) -> ExecuteResponse:
        """Execute a command in the sandbox."""
        if not self._container:
            return ExecuteResponse(
                output="Container not running",
                exit_code=1,
                truncated=False
            )

        try:
            # Docker exec_run 返回 (exit_code, output)
            # output 是字节类型
            # 使用列表形式的 cmd 以避免 shell 转义问题
            exit_code, output = self._container.exec_run(
                cmd=["bash", "-c", command],
                workdir=self.working_dir,
                demux=False # Combine stdout and stderr
            )

            return ExecuteResponse(
                output=output.decode("utf-8", errors="replace"),
                exit_code=exit_code,
                truncated=False,
            )
        except Exception as e:
            return ExecuteResponse(
                output=f"Error executing command: {str(e)}",
                exit_code=1,
                truncated=False,
            )

    def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]:
        """Upload multiple files to the sandbox using tar archive."""
        if not self._container:
            return [FileUploadResponse(path=p, error="permission_denied") for p, _ in files]

        responses = []

        # Create a tar archive in memory
        tar_stream = io.BytesIO()
        with tarfile.open(fileobj=tar_stream, mode='w') as tar:
            for path, content in files:
                # Docker put_archive expects relative paths inside the tar to be relative to the destination
                # But here we want absolute paths to be respected.
                # Actually put_archive extracts to a directory.
                # To support absolute paths, we should probably upload to root /?
                # Or handle relative paths relative to working_dir.

                # Let's handle paths:
                # If path is absolute, we strip leading / and upload to root.
                # If path is relative, we upload to working_dir.

                # Simplification: We will create a tar with full structure and extract to /

                # Normalize path
                if path.startswith("/"):
                    arcname = path.lstrip("/")
                    dest_path = "/"
                else:
                    arcname = path
                    dest_path = self.working_dir

                info = tarfile.TarInfo(name=arcname)
                info.size = len(content)
                info.mtime = time.time()
                tar.addfile(info, io.BytesIO(content))

                responses.append(FileUploadResponse(path=path, error=None))

        tar_stream.seek(0)

        try:
            # We extract to / to support absolute paths in the tar
            # Note: This assumes all files in the batch can be extracted to the same root.
            # If mixed absolute/relative, this might be tricky.
            # For robustness, we might need to upload one by one if paths are mixed,
            # or group them.
            # Strategy: Always extract to / (root), and ensure arcnames are full paths (without leading /)

            self._container.put_archive(
                path="/",
                data=tar_stream
            )
        except Exception as e:
            # Mark all as failed if batch fails
            return [FileUploadResponse(path=p, error="permission_denied") for p, _ in files]

        return responses

    def download_files(self, paths: list[str]) -> list[FileDownloadResponse]:
        """Download multiple files from the sandbox."""
        if not self._container:
            return [FileDownloadResponse(path=p, error="permission_denied") for p in paths]

        responses = []
        for path in paths:
            try:
                # get_archive returns a tuple (generator, stat)
                bits, stat = self._container.get_archive(path)

                # Reconstruct tar from bits
                file_content = io.BytesIO()
                for chunk in bits:
                    file_content.write(chunk)
                file_content.seek(0)

                # Extract file from tar
                with tarfile.open(fileobj=file_content, mode='r') as tar:
                    # There should be only one file/dir
                    member = tar.next()
                    if member.isdir():
                        responses.append(FileDownloadResponse(path=path, error="is_directory"))
                        continue

                    f = tar.extractfile(member)
                    if f:
                        content = f.read()
                        responses.append(FileDownloadResponse(path=path, content=content, error=None))
                    else:
                        responses.append(FileDownloadResponse(path=path, error="file_not_found"))

            except NotFound:
                responses.append(FileDownloadResponse(path=path, error="file_not_found"))
            except Exception as e:
                error_msg = str(e).lower()
                error = "invalid_path"
                if "permission" in error_msg:
                    error = "permission_denied"

                responses.append(FileDownloadResponse(path=path, content=None, error=error))
        return responses

    def close(self):
        """Close the sandbox session."""
        if self._container:
            try:
                if self.auto_remove:
                    self._container.remove(force=True)
                else:
                    self._container.stop()
            except Exception:
                pass
            self._container = None