from dotenv import load_dotenv
from openai import OpenAI

load_dotenv(dotenv_path='../../.env')

client = OpenAI()

response = client.responses.create(model="gpt-4o-mini", input="Tell a quick dad joke!")

print(response)

response.to_dict()

print(response.output[0].content[0].text)

Or a little shortcut:

print(response.output_text)

client.responses.retrieve(response.id)

Instruction Following

response = client.responses.create(model="gpt-4o-mini", instructions="You return markdown and lots of emojis. ", input="Tell a quick dad joke!")
print(response.output_text)

response = client.responses.create(
    model="gpt-4o-mini",
    input=[{"role": "developer", "content": "You return markdown and lots of emojis. "}, {"role": "user", "content": "Tell a quick dad joke!"}],
)

print(response.output_text)

The argument instructions is used to insert a system (or developer) message as the first item in the model's context source.

response = client.responses.create(
    model="gpt-4o-mini",
    input=[{"role": "developer", "content": "You return markdown and lots of emojis. "}, {"role": "user", "content": "Tell a quick dad joke!"}],
    instructions="You must talk like a pirate and do not return any markdown or emojis.",
)

print(response.output_text)

response = client.responses.create(
    model="gpt-4o-mini",
    input=[
        {"role": "developer", "content": "You are a helpful assistant."},
        {"role": "user", "content": "My name is Chris, and my age is 40."},
        {"role": "assistant", "content": "Nice to meet you, Chris!"},
        {"role": "user", "content": "How old am I?"},
    ],
)

print(response.output_text)

Alternately, we can use the previous_response_id parameter to manage conversation state.

response = client.responses.create(
    model="gpt-4o-mini",
    input=[
        {"role": "developer", "content": "You are a helpful assistant."},
        {"role": "user", "content": "My name is Chris, and my age is 40."},
    ],
)

print(response.output_text)
print(response.id)

response = client.responses.create(
    model="gpt-4o-mini",
    input=[
        {"role": "user", "content": "How old am I"},
    ],
    previous_response_id=response.id,
)
print(response.output_text)
print(response.id)

response = client.responses.create(
    model="gpt-4o-mini",
    input=[
        {"role": "user", "content": "And what was my name?"},
    ],
    previous_response_id=response.id,
)
print(response.output_text)
print(response.id)

When using previous_response_id, the instructions from a previous response will be not be carried over to the next response. This makes it simple to swap out system (or developer) messages in new responses source. The instructions parameter only applies to the current response generation request.

response = client.responses.create(
    model="gpt-4o-mini",
    input=[
        {"role": "user", "content": "What country is the city Toronto in?"},
    ],
    instructions="You only write lower case letters",
)
print(response.output_text)
response.id

response = client.responses.create(
    model="gpt-4o-mini",
    input=[
        {"role": "user", "content": "What country was it again?"},
    ],
    previous_response_id=response.id,
)
print(response.output_text)

response = client.responses.create(
    model="gpt-4o-mini",
    input=[
        {"role": "user", "content": "I forget, what was it?"},
    ],
    instructions="You only write UPPER CASE letters",
    previous_response_id=response.id,
)
print(response.output_text)

# Importing these to make the output look nicer
from fasthtml.common import show
from monsterui.all import render_md

Web Search

response = client.responses.create(
    model="gpt-4o-mini",
    tools=[{"type": "web_search_preview"}],
    input="Did Alabama State win their first game of the March Madness tournament in 2025?",  # web_search_preview_2025_03_11 points to a dated version of the tool
)

show(render_md(response.output_text))

response.to_dict()

You can also force the use of the web_search_preview tool by using the tool_choice parameter, and setting it to {type: "web_search_preview"}:

response = client.responses.create(
    model="gpt-4o-mini",
    tools=[{"type": "web_search_preview"}],
    tool_choice={"type": "web_search_preview"},
    input="Is there an upcoming federal election in Canada?",
)
show(render_md(response.output_text))

response.to_dict()

response.output[0]

• the annotations:

response.output[1].content[0].annotations

• text content:

response.output[1].content[0].text

To refine search results based on geography, you can specify an approximate user location using country, city, region, and/or timezone.

response = client.responses.create(
    model="gpt-4o-mini",
    tools=[
        {
            "type": "web_search_preview",
            "user_location": {
                "type": "approximate",
                "country": "CA",  #  two-letter ISO country code
                "city": "Halifax",  # free text strings
                "region": "Nova Scotia",  # free text strings
            },
        }
    ],
    input="What are the best restaurants around Halifax?",
)

show(render_md(response.output_text))

The parameter search_context_size controls the number of search results. The tokens used by the search tool do not affect the context window of the main model. Choosing the search_context_size parameter is a trade-off between cost, quality, and latency. The available values are 'high', 'medium', and 'low'. The default is 'medium'. The pricing page has all the details

response = client.responses.create(
    model="gpt-4o",
    tools=[
        {
            "type": "web_search_preview",
            "user_location": {
                "type": "approximate",
            },
            "search_context_size": "high",
        }
    ],
    input="Give me a markdown table of all the 2025 March Madness games and scores that have been played so far.",
)

show(render_md(response.output_text))

File Search

from agents import Agent, Runner

agent = Agent(name="Chris", instructions="You are a helpful assistant.")
result = await Runner.run(agent, "Hey")
print(result.final_output)

result.to_input_list()

result

import json

import modal

# Create image with IPython installed
image = modal.Image.debian_slim().pip_install("ipython", "pandas")


# Create the driver program that will run in the sandbox
def create_driver_program():
    return """
import json
import sys
import re
from IPython.core.interactiveshell import InteractiveShell
from IPython.utils.io import capture_output

def strip_ansi_codes(text):
    ansi_escape = re.compile(r'\\x1B(?:[@-Z\\\\-_]|\\[[0-?]*[ -/]*[@-~])')
    return ansi_escape.sub('', text)

# Create a persistent IPython shell instance
shell = InteractiveShell()
shell.colors = 'NoColor'  # Disable color output
shell.autoindent = False  # Disable autoindent

# Keep reading commands from stdin
while True:
    try:
        # Read a line of JSON from stdin
        command = json.loads(input())
        code = command.get('code')
        
        if code is None:
            print(json.dumps({"error": "No code provided"}))
            continue
            
        # Execute the code and capture output
        with capture_output() as captured:
            result = shell.run_cell(code)

        # Clean the outputs
        stdout = strip_ansi_codes(captured.stdout)
        stderr = strip_ansi_codes(captured.stderr)
        error = strip_ansi_codes(str(result.error_in_exec)) if not result.success else None

        # Format the response
        response = {
            "stdout": stdout,
            "stderr": stderr,
            "success": result.success,
            "result": repr(result.result) if result.success else None,
            "error": error
        }
        
        # Send the response
        print(json.dumps(response), flush=True)
        
    except Exception as e:
        print(json.dumps({"error": strip_ansi_codes(str(e))}), flush=True)
"""


def create_sandbox(timeout=300):
    """Creates and returns a Modal sandbox running an IPython shell."""
    app = modal.App.lookup("ipython-sandbox", create_if_missing=True)

    # Create the sandbox with the driver program
    with modal.enable_output():
        sandbox = modal.Sandbox.create("python", "-c", create_driver_program(), image=image, app=app, timeout=timeout)

    return sandbox


def execute_python_code(code: str, sandbox=None) -> dict:
    created_sandbox = False
    if sandbox is None:
        sandbox = create_sandbox()
        created_sandbox = True
    # Send the code to the sandbox
    sandbox.stdin.write(json.dumps({"code": code}))
    sandbox.stdin.write("\n")
    sandbox.stdin.drain()

    # Get the response
    response = next(iter(sandbox.stdout))
    if created_sandbox:
        sandbox.terminate()
    return json.loads(response)

I will make it a tool that can be used by the agent. We will also add the WebSearchTool to the agent as well.

from agents import WebSearchTool, function_tool

sb = create_sandbox()


@function_tool()
def execute_code(code: str) -> dict:
    """Execute the given Python code in a sandboxed environment and return the output."""
    return execute_python_code(code, sb)

coding_instructions = """
You solve tasks using an agentic coding loop in Python.

Follow this loop carefully:

- Think and write code. Send the code to the execute_code tool.
- Get code output results back.
- Think again, improve or add more code.
- Get code output results back.
...
- Repeat until you've solved the problem completely.

If you encounter any errors:

- FIX THEM and continue the loop.
- If modules are not found, install them using: !pip install <package_name>

Never give up. Continue iterating until the task is fully solved.
The sandbox environment is safe, isolated, and supports running arbitrary Python code.
State is maintained between code snippets:
Variables and definitions persist across executions.
# first code snippet
x = 2
print(x)
# second code snippet in separate request
y = 6
print(x + y)  # This works because state persists

Begin your agentic coding loop!

"""

coding_agent = Agent(
    name="Code Agent",
    instructions=coding_instructions,
    model="gpt-4o",
    tools=[execute_code, WebSearchTool()],
)

result = await Runner.run(
    coding_agent,
    """
                          Use an open source weather API to get the temperature in Halifax, Nova Scotia for the last 60 days. 
                          Then fit a statistical model to predict the temperature for the next week. 
                          Print the final predictions in a table (markdown format).
                          Todays date is 2025-04-02.""",
    max_turns=20,
)
print(result.final_output)

There are some different attributes that can be accessed on the result object. Here are some others worth noting:

print(result.input)

for r in result.raw_responses:
    print(r)

result.new_items

result.to_input_list()

We could hack together a function to display the conversation like this:

def show_conversation(conversation):
    from ast import literal_eval

    for message in conversation:
        if message.get("role") == "user":
            show(render_md("### User:"))
            show(render_md(message.get("content", "")))

        elif message.get("role") == "assistant":
            show(render_md("### Assistant:"))
            # Handle the new content format which is a list of dictionaries
            if isinstance(message.get("content"), list):
                for content_item in message["content"]:
                    if content_item.get("type") == "output_text":
                        show(render_md(content_item.get("text", "")))

        # Handle tool calls (like execute_code)
        elif message.get("type") == "function_call":
            show(render_md(f"### Tool Call: {message.get('name')}"))
            if "arguments" in message:
                arguments = json.loads(message["arguments"])
                for k, v in arguments.items():
                    show(render_md(f"#### Argument: {k}"))
                    show(render_md(f"```python\n{v}\n```"))

        # Handle tool outputs
        elif message.get("type") == "function_call_output":
            output = message.get("output", {})
            output = literal_eval(output)
            for k, v in output.items():
                if not v:
                    continue
                show(render_md(f"#### Output: {k}"))
                show(render_md(f"```python\n{v}\n```"))
        elif message.get("type") == "web_search_call":
            show(render_md("### Performing Web Search:"))
        else:
            raise Exception(f"Unknown message type: {message.get('type')}")


# Use it like this:
show_conversation(result.to_input_list())

from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX

print(RECOMMENDED_PROMPT_PREFIX)

review_agent = Agent(
    name="Review Agent",
    instructions=f"{RECOMMENDED_PROMPT_PREFIX} You need to review the code, logic, and answer provided by the coding agent. If the coding agent does not provide a complete answer then explain to the coding agent what it needs to do to fix it.",
    model="gpt-4o",
    handoff_description="Specialist agent for reviewing code",
    handoffs=[coding_agent],
)

coding_agent = Agent(
    name="Code Agent",
    instructions=f"{RECOMMENDED_PROMPT_PREFIX} {coding_instructions}. After completing your task, hand off the results to the review agent for final review.",
    model="gpt-4o",
    tools=[execute_code, WebSearchTool()],
    handoff_description="Specialist agent for coding and executing code",
    handoffs=[review_agent],
)

triage_agent = Agent(
    name="Triage Agent",
    instructions=f"{RECOMMENDED_PROMPT_PREFIX} You determine which agent to use based on the users request. You start by handing off the request and details to the coding agent.",
    model="gpt-4o",
    handoff_description="Specialist agent for triaging tasks",
    handoffs=[coding_agent, review_agent],
)

result = await Runner.run(
    triage_agent,
    "I need a table of dow jones closing prices for the last 30 days. Present the results in markdown format. Also compute the summary statistics of the closing prices using Python code.",
)
print(result.final_output)

result.to_input_list()

show_conversation(result.to_input_list())

from pydantic import BaseModel


class ReviewResult(BaseModel):
    success: bool
    feedback: str


review_agent = Agent(
    name="Review Agent",
    instructions="""
    Review the work done by the coding agent. 
    Give it feedback on whether or not it was successful.
    Review the code and the output as well as the final answer to make sure it completely answers the user's request.
    If it does not, ask the coding agent to fix it.
    Note: The end user has no ability to view plots so always give feedback that results must be presented in markdown formatted tables instead.
    """,
    model="gpt-4o",
    output_type=ReviewResult,
)

coding_agent = Agent(
    name="Code Agent",
    instructions=coding_instructions,
    model="gpt-4o",
    tools=[execute_code, WebSearchTool()],
)

task = "I need data for dow jones closing prices for the last 30 days as well as a prediction for the next week. Plot the results in a line chart."
res = await Runner.run(coding_agent, task)
for i in range(5):
    res = await Runner.run(review_agent, res.to_input_list(), max_turns=20)
    if res.final_output.success:
        break
    else:
        res = await Runner.run(coding_agent, res.to_input_list(), max_turns=20)

res.to_input_list()

show_conversation(res.to_input_list())