┌─────────────────────────────────────────────────────────────────┐

│                     Agent Framework                              │

│            (Successor to AutoGen + Semantic Kernel)             │

│                                                                  │

│  ┌─────────────────────────────────────────────────────────┐    │

│  │  From AutoGen:                                           │    │

│  │  • Simple multi-agent conversations                      │    │

│  │  • Agent orchestration patterns                          │    │

│  │  • Group chat abstractions                               │    │

│  └─────────────────────────────────────────────────────────┘    │

│                              +                                   │

│  ┌─────────────────────────────────────────────────────────┐    │

│  │  From Semantic Kernel:                                   │    │

│  │  • Enterprise-grade state management                     │    │

│  │  • Thread-based conversations                            │    │

│  │  • Type safety and observability                         │    │

│  │  • Azure AI Foundry integration                          │    │

│  └─────────────────────────────────────────────────────────┘    │

│                              +                                   │

│  ┌─────────────────────────────────────────────────────────┐    │

│  │  New in Agent Framework:                                 │    │

│  │  • Workflow graphs for explicit execution paths          │    │

│  │  • Human-in-the-loop patterns                            │    │

│  │  • A2A (Agent-to-Agent) connections                      │    │

│  │  • MCP (Model Context Protocol) support                  │    │

│  └─────────────────────────────────────────────────────────┘    │

└─────────────────────────────────────────────────────────────────┘

pip install agent-framework hopx-ai

export OPENAI_API_KEY="sk-..."

export HOPX_API_KEY="..."

# Or for Azure OpenAI:

export AZURE_OPENAI_API_KEY="..."

export AZURE_OPENAI_ENDPOINT="https://..."

from agent_framework import ChatAgent, Tool, tool

from hopx import Sandbox

from typing import Optional

class HopXExecutor:

    """Manage HopX sandbox for code execution."""

    def __init__(self, persist: bool = True, ttl: int = 600):

        self.persist = persist

        self.ttl = ttl

        self._sandbox: Optional[Sandbox] = None

    @property

    def sandbox(self) -> Sandbox:

        """Get or create sandbox."""

        if self._sandbox is None:

            self._sandbox = Sandbox.create(

                template="code-interpreter",

                ttl=self.ttl if self.persist else 60

            )

        return self._sandbox

    def execute(self, code: str, language: str = "python") -> str:

        """Execute code in sandbox."""

        try:

            result = self.sandbox.runCode(code, language=language, timeout=60)

            if result.exitCode == 0:

                return result.stdout or "Code executed successfully (no output)"

            else:

                return f"Error (exit {result.exitCode}):\n{result.stderr}"

        except Exception as e:

            self._sandbox = None  # Reset on error

            return f"Execution failed: {str(e)}"

    def cleanup(self):

        """Destroy sandbox."""

        if self._sandbox:

            self._sandbox.kill()

            self._sandbox = None

# Create global executor

executor = HopXExecutor(persist=True)

@tool

def execute_python(code: str) -> str:

    """

    Execute Python code in a secure, isolated sandbox.

    Use this tool when you need to:

    - Perform calculations or mathematical operations

    - Analyze data with pandas, numpy

    - Create visualizations with matplotlib

    - Process files or data structures

    - Run any Python computation

    The sandbox has pandas, numpy, matplotlib, seaborn, scikit-learn installed.

    Always use print() to show results.

    State persists between calls within the same session.

    Args:

        code: Complete, executable Python code

    Returns:

        Output from code execution or error message

    """

    return executor.execute(code, "python")

@tool

def execute_bash(command: str) -> str:

    """

    Execute bash commands in a secure sandbox.

    Use for:

    - File operations (ls, cat, head)

    - Package installation (pip install)

    - System commands

    Args:

        command: Bash command to execute

    Returns:

        Command output or error

    """

    return executor.execute(command, "bash")

from agent_framework import ChatAgent, ChatClient

from agent_framework.models import OpenAIChatClient

import os

# Create chat client

client = OpenAIChatClient(

    model="gpt-4o",

    api_key=os.environ["OPENAI_API_KEY"]

)

# Or for Azure OpenAI:

# from agent_framework.models import AzureOpenAIChatClient

# client = AzureOpenAIChatClient(

#     deployment_name="gpt-4o",

#     endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],

#     api_key=os.environ["AZURE_OPENAI_API_KEY"]

# )

# Create agent with tools

agent = ChatAgent(

    name="CodeAssistant",

    client=client,

    tools=[execute_python, execute_bash],

    system_message="""You are a helpful AI assistant that can execute Python code.

When users ask questions that require computation:

1. Write clear, well-documented Python code

2. Use the execute_python tool to run it

3. Analyze the results and explain them

Available libraries: pandas, numpy, matplotlib, seaborn, scipy, scikit-learn.

For visualizations, save to /app/chart.png using plt.savefig().

Always print() results you want to see.

"""

)

# Get a new conversation thread

thread = agent.get_new_thread()

# Chat with the agent

response = await agent.chat(

    thread=thread,

    message="Calculate the first 50 prime numbers and find their sum"

)

print(response.content)

# Continue the conversation (state persists)

response = await agent.chat(

    thread=thread,

    message="Now find the average and standard deviation of those primes"

)

print(response.content)

# Cleanup

executor.cleanup()

from agent_framework import ChatAgent, Workflow, Node, Edge

from agent_framework.models import OpenAIChatClient

# Create specialized agents

researcher = ChatAgent(

    name="Researcher",

    client=client,

    system_message="You research topics and identify key questions to answer."

)

analyst = ChatAgent(

    name="Analyst",

    client=client,

    tools=[execute_python],

    system_message="You analyze data using Python. Always verify with code."

)

writer = ChatAgent(

    name="Writer",

    client=client,

    system_message="You create clear, well-structured reports from analysis results."

)

# Define workflow graph

workflow = Workflow(

    name="ResearchPipeline",

    nodes=[

        Node(id="research", agent=researcher),

        Node(id="analyze", agent=analyst),

        Node(id="report", agent=writer)

    ],

    edges=[

        Edge(source="research", target="analyze"),

        Edge(source="analyze", target="report")

    ]

)

# Execute workflow

result = await workflow.run(

    input="Analyze the relationship between company size and revenue growth"

)

print(result.output)

from agent_framework import ChatAgent, tool

from hopx import Sandbox

from typing import Optional

import json

class SharedSandboxManager:

    """Singleton sandbox shared across all agents."""

    _instance: Optional['SharedSandboxManager'] = None

    _sandbox: Optional[Sandbox] = None

    @classmethod

    def get_instance(cls) -> 'SharedSandboxManager':

        if cls._instance is None:

            cls._instance = cls()

        return cls._instance

    def get_sandbox(self) -> Sandbox:

        if self._sandbox is None:

            self._sandbox = Sandbox.create(

                template="code-interpreter",

                ttl=900  # 15 minutes for long workflows

            )

            # Initialize common imports

            self._sandbox.runCode("""

import pandas as pd

import numpy as np

import matplotlib.pyplot as plt

import seaborn as sns

import json

from datetime import datetime

print("Shared environment ready!")

""", language="python", timeout=30)

        return self._sandbox

    def execute(self, code: str) -> str:

        sandbox = self.get_sandbox()

        result = sandbox.runCode(code, language="python", timeout=60)

        return result.stdout if result.exitCode == 0 else f"Error: {result.stderr}"

    def cleanup(self):

        if self._sandbox:

            self._sandbox.kill()

            self._sandbox = None

@tool

def shared_python(code: str) -> str:

    """

    Execute Python with SHARED STATE across all agents.

    Variables, DataFrames, and computations persist between calls

    from ANY agent in the workflow. Use for collaborative analysis.

    Args:

        code: Python code to execute

    Returns:

        Execution output

    """

    return SharedSandboxManager.get_instance().execute(code)

# Create agents that share state

data_engineer = ChatAgent(

    name="DataEngineer",

    client=client,

    tools=[shared_python],

    system_message="You prepare and clean data. Save results to variables for other agents."

)

data_scientist = ChatAgent(

    name="DataScientist",

    client=client,

    tools=[shared_python],

    system_message="You analyze prepared data. Build on variables created by DataEngineer."

)

# In a workflow, both agents access the same sandbox

# DataEngineer: df = pd.read_csv(...)

# DataScientist: df['new_col'] = ...  # Same df!

from agent_framework import ChatAgent, HumanApprovalHandler

from agent_framework.handlers import ConsoleApprovalHandler

# Custom approval handler for code execution

class CodeApprovalHandler(HumanApprovalHandler):

    """Require human approval before executing code."""

    async def request_approval(self, action: str, details: dict) -> bool:

        if action == "tool_call" and details.get("tool") == "execute_python":

            code = details.get("arguments", {}).get("code", "")

            print("\n" + "="*60)

            print("🔍 Code Execution Request")

            print("="*60)

            print(code)

            print("="*60)

            response = input("Approve execution? [y/N]: ").strip().lower()

            return response == "y"

        return True  # Auto-approve other actions

# Create agent with approval handler

secure_agent = ChatAgent(

    name="SecureAssistant",

    client=client,

    tools=[execute_python],

    approval_handler=CodeApprovalHandler()

)

# Now code execution requires human approval

response = await secure_agent.chat(

    thread=thread,

    message="Delete all files in /app"  # This will prompt for approval!

)

# AutoGen way

from autogen import AssistantAgent, UserProxyAgent

assistant = AssistantAgent(

    name="assistant",

    llm_config={"config_list": [{"model": "gpt-4o"}]}

)

user_proxy = UserProxyAgent(

    name="user_proxy",

    code_execution_config={"executor": docker_executor}

)

user_proxy.initiate_chat(assistant, message="Hello")

# Agent Framework way

from agent_framework import ChatAgent

from agent_framework.models import OpenAIChatClient

agent = ChatAgent(

    name="assistant",

    client=OpenAIChatClient(model="gpt-4o"),

    tools=[execute_python]  # HopX-backed tool

)

thread = agent.get_new_thread()

response = await agent.chat(thread=thread, message="Hello")

"""

Microsoft Agent Framework with HopX Code Execution

"""

import asyncio

import os

from typing import Optional

from agent_framework import ChatAgent, tool, Workflow, Node, Edge

from agent_framework.models import OpenAIChatClient

from hopx import Sandbox

# Verify environment

assert os.environ.get("OPENAI_API_KEY"), "Set OPENAI_API_KEY"

assert os.environ.get("HOPX_API_KEY"), "Set HOPX_API_KEY"

class SandboxManager:

    """Manage shared HopX sandbox."""

    _sandbox: Optional[Sandbox] = None

    @classmethod

    def execute(cls, code: str, language: str = "python") -> str:

        if cls._sandbox is None:

            cls._sandbox = Sandbox.create(template="code-interpreter", ttl=600)

        result = cls._sandbox.runCode(code, language=language, timeout=60)

        return result.stdout if result.exitCode == 0 else f"Error: {result.stderr}"

    @classmethod

    def cleanup(cls):

        if cls._sandbox:

            cls._sandbox.kill()

            cls._sandbox = None

@tool

def python(code: str) -> str:

    """Execute Python code. State persists between calls."""

    return SandboxManager.execute(code, "python")

@tool

def bash(command: str) -> str:

    """Execute bash commands."""

    return SandboxManager.execute(command, "bash")

async def main():

    # Create client

    client = OpenAIChatClient(

        model="gpt-4o",

        api_key=os.environ["OPENAI_API_KEY"]

    )

    # Create agent

    agent = ChatAgent(

        name="DataAnalyst",

        client=client,

        tools=[python, bash],

        system_message="""You are a data analyst. Use Python for all calculations.

Always verify results with code. Available: pandas, numpy, matplotlib."""

    )

    # Start conversation

    thread = agent.get_new_thread()

    try:

        # Multi-turn conversation with persistent state

        print("Query 1: Creating data...")

        r1 = await agent.chat(

            thread=thread,

            message="Create a DataFrame with 500 rows of sales data: date, product, region, amount"

        )

        print(r1.content)

        print("\nQuery 2: Analyzing...")

        r2 = await agent.chat(

            thread=thread,

            message="Calculate total sales by region and find the top performer"

        )

        print(r2.content)

        print("\nQuery 3: Visualization...")

        r3 = await agent.chat(

            thread=thread,

            message="Create a bar chart of sales by region"

        )

        print(r3.content)

    finally:

        SandboxManager.cleanup()

        print("\n✅ Sandbox cleaned up")

if __name__ == "__main__":

    asyncio.run(main())

async def run_research_pipeline():

    """Run a multi-agent research workflow."""

    client = OpenAIChatClient(model="gpt-4o")

    # Specialized agents

    researcher = ChatAgent(

        name="Researcher",

        client=client,

        system_message="Identify key research questions and data requirements."

    )

    analyst = ChatAgent(

        name="Analyst", 

        client=client,

        tools=[python],

        system_message="Perform data analysis. Always use code for accuracy."

    )

    writer = ChatAgent(

        name="Writer",

        client=client,

        tools=[python],

        system_message="Create clear reports. Verify numbers with code."

    )

    # Define workflow

    workflow = Workflow(

        name="MarketResearch",

        nodes=[

            Node(id="research", agent=researcher),

            Node(id="analyze", agent=analyst),

            Node(id="report", agent=writer)

        ],

        edges=[

            Edge(source="research", target="analyze"),

            Edge(source="analyze", target="report")

        ]

    )

    try:

        result = await workflow.run(

            input="""

            Research Question: What factors correlate with high customer lifetime value?

            Create synthetic customer data and analyze:

            - Demographics (age, income, location)

            - Behavior (purchase frequency, avg order value)

            - Engagement (email opens, support tickets)

            Identify the top 3 predictive factors.

            """

        )

        print("="*60)

        print("RESEARCH REPORT")

        print("="*60)

        print(result.output)

    finally:

        SandboxManager.cleanup()

asyncio.run(run_research_pipeline())

# Agent Framework is async-first

response = await agent.chat(thread=thread, message=query)

# Create thread for conversation

thread = agent.get_new_thread()

# Use same thread for multi-turn

r1 = await agent.chat(thread=thread, message="Step 1")

r2 = await agent.chat(thread=thread, message="Step 2")  # Continues context

try:

    result = await workflow.run(input=task)

finally:

    SandboxManager.cleanup()

@tool

def safe_python(code: str) -> str:

    """Execute Python with error handling."""

    try:

        return SandboxManager.execute(code)

    except Exception as e:

        SandboxManager._sandbox = None  # Reset on failure

        return f"Execution failed: {e}. Please try again."