Using OpenAI Mode¶

This notebook walks you through enabling OpenAI mode to serve a PydanticAI agent, and connecting to it from the OpenAI SDK.

To run this notebook, make sure you have fastapi, fastapi-agents, pydantic-ai and openai installed (e.g. pip install fastapi fastapi-agents pydantic-ai openai).

Create your agent with FastAPI Agents¶

We're going to use a simple weather agent for the demo. The weather is always great!

In [1]:

from fastapi_agents import FastAPIAgents
from fastapi_agents.pydantic_ai import PydanticAIAgent
from pydantic_ai import Agent

weather_agent = Agent("openai:gpt-4o")

@weather_agent.tool
async def get_weather(_, location: str) -> str:
    """Check the weather for a given location.
    
    Args:
        location (str): The city to check the weather
    """
    return f"The weather is great in {location}"

# IMPORTANT: mode = 'openai'
app = FastAPIAgents.as_app(mode='openai')
app.register("weather", PydanticAIAgent(weather_agent))

from fastapi_agents import FastAPIAgents from fastapi_agents.pydantic_ai import PydanticAIAgent from pydantic_ai import Agent weather_agent = Agent("openai:gpt-4o") @weather_agent.tool async def get_weather(_, location: str) -> str: """Check the weather for a given location. Args: location (str): The city to check the weather """ return f"The weather is great in {location}" # IMPORTANT: mode = 'openai' app = FastAPIAgents.as_app(mode='openai') app.register("weather", PydanticAIAgent(weather_agent))

Run our API in the background¶

Since we're in Jupyter, we're going to run our API as a thread so we can execute the cells following it. We also need to nest asyncio since Jupyter is already running its own asyncio event loop.

In [ ]:

import nest_asyncio
import threading
from uvicorn import Config, Server

# Apply nest_asyncio to handle event loop issues in Jupyter
nest_asyncio.apply()

# Define a server instance
config = Config(app=app, log_level="info")
server = Server(config)

# Function to run the server
def start_server():
    server.run()

# Start the server in a background thread
thread = threading.Thread(target=start_server, daemon=True)
thread.start()

print("Uvicorn server is running in the background")

import nest_asyncio import threading from uvicorn import Config, Server # Apply nest_asyncio to handle event loop issues in Jupyter nest_asyncio.apply() # Define a server instance config = Config(app=app, log_level="info") server = Server(config) # Function to run the server def start_server(): server.run() # Start the server in a background thread thread = threading.Thread(target=start_server, daemon=True) thread.start() print("Uvicorn server is running in the background")

Connecting OpenAI SDK¶

The OpenAI SDK allows us to change the base_url for requests, so we're going to use the FastAPI local server we just started.

Let's check our the agent we registered is available!

In [ ]:

from openai import OpenAI

client = OpenAI(base_url=f'http://{server.config.host}:{server.config.port}')

for model in client.models.list():
    model_id = model.id
    break

print(f"Found {model_id}")

from openai import OpenAI client = OpenAI(base_url=f'http://{server.config.host}:{server.config.port}') for model in client.models.list(): model_id = model.id break print(f"Found {model_id}")

Create a chat completion¶

We can interact with our agent as if it were an OpenAI model by using the chat completions API.

Just specify the correct model ID and messages as usual.

Please note that OpenAI supports many different functions in their official API, most of which are not currently available in FastAPI Agents.

If you have a specific request, please open an issue on GitHub with the details.

In [ ]:

response = client.chat.completions.create(
    model=model_id,
    messages=[{
        "role": "user",
        "content": "what is the weather in New York?"
    }]
)

print(response.choices[0].message.content)

response = client.chat.completions.create( model=model_id, messages=[{ "role": "user", "content": "what is the weather in New York?" }] ) print(response.choices[0].message.content)

Cleaning up¶

When we're done, we can easily stop the server we were running in the background.

In [ ]:

# Stop the server
server.should_exit = True
thread.join()  # Wait for the thread to finish

print("Uvicorn server has been stopped")

# Stop the server server.should_exit = True thread.join() # Wait for the thread to finish print("Uvicorn server has been stopped")