Documentation

OpenAI Agents SDK + ModelRiver

Run OpenAI's official agent framework through ModelRiver. Get provider-agnostic execution, automatic failover, and full observability for every agent action.

Overview

The OpenAI Agents SDK provides a lightweight, production-ready framework for building autonomous agents with tool use, handoffs between agents, and input/output guardrails. By pointing it at ModelRiver, your agents can use any provider: not just OpenAI: with automatic failover.

What you get:

  • Run the Agents SDK against any provider (OpenAI, Anthropic, Mistral, etc.)
  • Automatic failover during multi-step agent loops
  • Per-step cost tracking in Request Logs
  • Provider switching from the console without code changes

Quick start

Install dependencies

Bash
pip install openai-agents

Connect to ModelRiver

PYTHON
1from agents import Agent, Runner, set_default_openai_api, set_default_openai_client, set_tracing_disabled
2from openai import AsyncOpenAI
3 
4client = AsyncOpenAI(
5 base_url="https://api.modelriver.com/v1",
6 api_key="mr_live_YOUR_API_KEY",
7)
8set_default_openai_client(client, use_for_tracing=False)
9set_default_openai_api("chat_completions") # ModelRiver has no Responses API
10set_tracing_disabled(True) # traces would otherwise be exported to OpenAI
11 
12agent = Agent(
13 name="Assistant",
14 instructions="You are a helpful assistant that answers questions clearly.",
15 model="my_chat_workflow",
16)
17 
18result = Runner.run_sync(agent, "What is ModelRiver?")
19print(result.final_output)

The Agents SDK calls the OpenAI Responses API by default, which ModelRiver doesn't support. set_default_openai_api("chat_completions") switches it to Chat Completions, and set_default_openai_client routes every agent through ModelRiver. Tracing is turned off because the SDK would otherwise export traces to OpenAI, which needs an OpenAI API key. Each agent's model is a workflow name.

Agent with tools

PYTHON
1from agents import Agent, Runner, function_tool, set_default_openai_api, set_default_openai_client, set_tracing_disabled
2from openai import AsyncOpenAI
3 
4client = AsyncOpenAI(
5 base_url="https://api.modelriver.com/v1",
6 api_key="mr_live_YOUR_API_KEY",
7)
8set_default_openai_client(client, use_for_tracing=False)
9set_default_openai_api("chat_completions") # ModelRiver has no Responses API
10set_tracing_disabled(True) # traces would otherwise be exported to OpenAI
11 
12@function_tool
13def get_weather(location: str) -> str:
14 """Get the current weather for a location."""
15 return f"22°C and sunny in {location}"
16 
17@function_tool
18def search_knowledge_base(query: str) -> str:
19 """Search the internal knowledge base."""
20 return f"Found: ModelRiver routes AI requests across providers."
21 
22agent = Agent(
23 name="Research Assistant",
24 instructions="Help users find information. Use tools when needed.",
25 model="my_chat_workflow",
26 tools=[get_weather, search_knowledge_base],
27)
28 
29result = Runner.run_sync(agent, "What's the weather in Tokyo?")
30print(result.final_output)

Agent handoffs

Build multi-agent systems where agents delegate to specialists:

PYTHON
1from agents import Agent, Runner, set_default_openai_api, set_default_openai_client, set_tracing_disabled
2from openai import AsyncOpenAI
3 
4client = AsyncOpenAI(
5 base_url="https://api.modelriver.com/v1",
6 api_key="mr_live_YOUR_API_KEY",
7)
8set_default_openai_client(client, use_for_tracing=False)
9set_default_openai_api("chat_completions") # ModelRiver has no Responses API
10set_tracing_disabled(True) # traces would otherwise be exported to OpenAI
11 
12# Specialist agents
13billing_agent = Agent(
14 name="Billing Specialist",
15 instructions="You handle billing questions. Be precise about pricing.",
16 model="my_chat_workflow",
17)
18 
19tech_agent = Agent(
20 name="Tech Support",
21 instructions="You handle technical questions. Provide code examples.",
22 model="my_chat_workflow",
23)
24 
25# Triage agent that delegates
26triage_agent = Agent(
27 name="Triage",
28 instructions="Route the user to the right specialist based on their question.",
29 model="my_chat_workflow",
30 handoffs=[billing_agent, tech_agent],
31)
32 
33result = Runner.run_sync(triage_agent, "How much does the Pro plan cost?")
34print(result.final_output)

Guardrails

Add input and output validation:

PYTHON
1from agents import Agent, Runner, InputGuardrail, OutputGuardrail, GuardrailFunctionOutput
2from pydantic import BaseModel
3 
4class SafetyCheck(BaseModel):
5 is_safe: bool
6 reason: str
7 
8async def check_input(ctx, agent, input_text):
9 # Use a fast workflow for guardrail checks
10 result = await Runner.run(
11 Agent(
12 name="Safety Checker",
13 instructions="Check if the input is safe and appropriate. Return is_safe=false for harmful content.",
14 model="fast_safety_check", # Cheap, fast workflow
15 output_type=SafetyCheck,
16 ),
17 input_text,
18 context=ctx,
19 )
20 return GuardrailFunctionOutput(
21 output_info=result.final_output,
22 tripwire_triggered=not result.final_output.is_safe,
23 )
24 
25agent = Agent(
26 name="Guarded Assistant",
27 instructions="You are a helpful assistant.",
28 model="my_chat_workflow",
29 input_guardrails=[InputGuardrail(guardrail_function=check_input)],
30)

Streaming

PYTHON
1import asyncio
2from agents import Agent, Runner, set_default_openai_api, set_default_openai_client, set_tracing_disabled
3from openai import AsyncOpenAI
4 
5client = AsyncOpenAI(
6 base_url="https://api.modelriver.com/v1",
7 api_key="mr_live_YOUR_API_KEY",
8)
9set_default_openai_client(client, use_for_tracing=False)
10set_default_openai_api("chat_completions") # ModelRiver has no Responses API
11set_tracing_disabled(True) # traces would otherwise be exported to OpenAI
12 
13agent = Agent(
14 name="Storyteller",
15 instructions="Tell creative, engaging stories.",
16 model="my_chat_workflow",
17)
18 
19async def main():
20 result = Runner.run_streamed(agent, "Tell me a story about AI")
21 async for event in result.stream_events():
22 if event.type == "raw_response_event" and hasattr(event.data, "delta"):
23 print(event.data.delta, end="", flush=True)
24 
25asyncio.run(main())

Per-agent model routing

Use different ModelRiver workflows for different agent roles:

PYTHON
1# Fast model for triage (speed matters)
2triage_agent = Agent(name="Triage", model="fast_triage", ...)
3 
4# Powerful model for complex answers
5expert_agent = Agent(name="Expert", model="deep_expert", ...)
6 
7# Cheap model for guardrails
8safety_agent = Agent(name="Safety", model="fast_safety_check", ...)

Best practices

  1. Use different workflows per agent: Triage agents need speed, experts need depth
  2. Configure guardrails with cheap models: Safety checks should use fast, inexpensive workflows
  3. Monitor handoff chains: Check Request Logs for multi-step costs
  4. Set max turns: Prevent runaway agent loops that generate excessive costs
  5. Use structured outputs: Define output schemas for agents that produce structured data

Next steps