数据分析师

import asyncio
from dedalus_labs import AsyncDedalus, DedalusRunner
from dedalus_labs.utils.stream import stream_async
from dotenv import load_dotenv

load_dotenv()

def execute_python_code(code: str) -> str:
    """
    执行 Python 代码并返回结果。
    在受控命名空间中安全地执行代码。
    """
    try:
        namespace = {}
        exec(code, {"__builtins__": __builtins__}, namespace)

        if 'result' in namespace:
            return str(namespace['result'])

        results = {k: v for k, v in namespace.items() if not k.startswith('_')}
        return str(results) if results else "Code executed successfully"
    except Exception as e:
        return f"Error executing code: {str(e)}"

async def main():
    client = AsyncDedalus()
    runner = DedalusRunner(client)

    result = runner.run(
        input="""Research the current stock price of Tesla (TSLA) and Apple (AAPL).
        Then write and execute Python code to:
        1. Compare their current prices
        2. Calculate the percentage difference
        3. Determine which stock has grown more in the past year based on the data you find
        4. Provide investment insights based on your analysis

        Use web search to get the latest stock information.""",
        model="openai/gpt-5",
        tools=[execute_python_code],
        mcp_servers=["windsor/brave-search-mcp"],
        stream=True
    )

    await stream_async(result)

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