[AI-517, AI-744] Add an OpenAI-compatible model provider sample for Baseten, Fireworks AI, and Together AI

openai_agents/README.md

@@ -36,7 +36,7 @@ Each directory contains a complete example with its own README for detailed inst
 - **[Hosted MCP](./hosted_mcp/README.md)** - Using the MCP client functionality of the OpenAI Responses API.
 - **[MCP v2](./mcp_v2/README.md)** - MCP SDK v2 using named server factories and durable workflow proxies.
 - **[MCP](./mcp/README.md)** - Legacy MCP provider examples retained for compatibility.
-- **[Model Providers](./model_providers/README.md)** - Using custom LLM providers (e.g., Anthropic via LiteLLM).
+- **[Model Providers](./model_providers/README.md)** - Using custom LLM providers: Anthropic via LiteLLM, OpenRouter, a local gpt-oss with Ollama, and open-weight models on Baseten, Fireworks AI, and Together AI.
 - **[Research Bot](./research_bot/README.md)** - Multi-agent research system with specialized roles: a planner agent, search agent, and writer agent working together to conduct comprehensive research.
 - **[Customer Service](./customer_service/README.md)** - Interactive customer service agent with escalation capabilities, demonstrating conversational workflows.
 - **[Reasoning Content](./reasoning_content/README.md)** - Example of how to retrieve the thought process of reasoning models.

openai_agents/model_providers/README.md

@@ -47,6 +47,29 @@ uv run openai_agents/model_providers/run_openrouter_workflow.py
 
 The workflow uses `openrouter/auto`, so OpenRouter picks a model per request; change `OPENROUTER_MODEL` in [workflows/openrouter_workflow.py](workflows/openrouter_workflow.py) to pin any OpenRouter model slug. Tools that run inside the Workflow must be `async`: the Agents SDK runs sync tools in a thread, which the Workflow sandbox does not allow. See the [openrouter](../../openrouter) sample for calling OpenRouter directly from Activities with cost tracking and budgets.
 
+#### Open-weight models on Baseten, Fireworks AI, and Together AI
+[Baseten](https://www.baseten.co/), [Fireworks AI](https://fireworks.ai/), and [Together AI](https://www.together.ai/) serve open-weight models behind the OpenAI Chat Completions API. One Worker serves all three: it builds a client for each provider whose API key is set, and each Workflow picks a provider through the model name it gives its agent, `<provider>/<model>`. Workflows on different providers run side by side on the same Worker.
+
+| Provider | API key | Example model name |
+|---|---|---|
+| Baseten | `BASETEN_API_KEY` | `baseten/openai/gpt-oss-120b` |
+| Fireworks AI | `FIREWORKS_API_KEY` | `fireworks/accounts/fireworks/models/gpt-oss-120b` |
+| Together AI | `TOGETHER_API_KEY` | `together/openai/gpt-oss-120b` |
+
+Export the keys you have and start the worker:
+```bash
+export TOGETHER_API_KEY="your_together_api_key"
+
+uv run openai_agents/model_providers/run_openai_compatible_worker.py
+```
+
+Then run the example in a separate terminal. Pass `--model` once per Workflow to start; with no `--model`, it starts one Workflow per provider whose key is set in that shell:
+```bash
+uv run openai_agents/model_providers/run_openai_compatible_workflow.py --model together/openai/gpt-oss-120b
+```
+
+Any model the provider serves works with its prefix. A model name for a provider without a key, or with an unknown prefix, fails its Workflow without retrying. The providers are defined in [openai_compatible_providers.py](openai_compatible_providers.py); add another OpenAI-compatible endpoint by appending to `PROVIDERS`. Tools that run inside the Workflow must be `async`: the Agents SDK runs sync tools in a thread, which the Workflow sandbox does not allow. The Temporal docs have a page per provider: [Baseten](https://docs.temporal.io/develop/python/integrations/baseten), [Fireworks AI](https://docs.temporal.io/develop/python/integrations/fireworks-ai), and [Together AI](https://docs.temporal.io/develop/python/integrations/together-ai).
+
 ### Extra
 
 #### GPT-OSS with Ollama

openai_agents/model_providers/openai_compatible_providers.py

@@ -0,0 +1,113 @@
+"""Open-weight model providers with OpenAI-compatible APIs.
+
+Baseten, Fireworks AI, and Together AI all serve open-weight models behind the
+OpenAI Chat Completions API, so the stock ``openai`` client reaches them once
+it is pointed at the provider's base URL. One Worker serves every provider
+whose API key is set, and each Workflow picks a provider through the model
+name it gives its agent: ``<provider>/<model>``, such as
+``together/openai/gpt-oss-120b``.
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Iterable
+from dataclasses import dataclass
+
+from agents import Model, ModelProvider, OpenAIChatCompletionsModel
+from openai import AsyncOpenAI
+from temporalio.exceptions import ApplicationError
+
+
+@dataclass(frozen=True)
+class OpenAICompatibleProvider:
+    """An OpenAI-compatible inference endpoint and an example model on it."""
+
+    # The prefix that selects this provider in a model name.
+    name: str
+    base_url: str
+    api_key_env: str
+    # A model with tool-calling support, in the provider's naming scheme. The
+    # starter runs it when no model is given.
+    default_model: str
+
+
+# @@@SNIPSTART python-openai-agents-baseten-provider
+BASETEN = OpenAICompatibleProvider(
+    name="baseten",
+    base_url="https://inference.baseten.co/v1",
+    api_key_env="BASETEN_API_KEY",
+    default_model="openai/gpt-oss-120b",
+)
+# @@@SNIPEND
+
+# @@@SNIPSTART python-openai-agents-fireworks-provider
+FIREWORKS = OpenAICompatibleProvider(
+    name="fireworks",
+    base_url="https://api.fireworks.ai/inference/v1",
+    api_key_env="FIREWORKS_API_KEY",
+    default_model="accounts/fireworks/models/gpt-oss-120b",
+)
+# @@@SNIPEND
+
+# @@@SNIPSTART python-openai-agents-together-provider
+TOGETHER = OpenAICompatibleProvider(
+    name="together",
+    base_url="https://api.together.ai/v1",
+    api_key_env="TOGETHER_API_KEY",
+    default_model="openai/gpt-oss-120b",
+)
+# @@@SNIPEND
+
+PROVIDERS: dict[str, OpenAICompatibleProvider] = {
+    provider.name: provider for provider in (BASETEN, FIREWORKS, TOGETHER)
+}
+
+
+# @@@SNIPSTART python-openai-agents-openai-compatible-model-provider
+class OpenAICompatibleModelProvider(ModelProvider):
+    """Routes ``<provider>/<model>`` names to the matching OpenAI-compatible endpoint.
+
+    One Worker serves every provider whose API key is set, so Workflows on
+    different providers run side by side on the same Task Queue. Client
+    retries are off: the plugin runs each model call as a Temporal Activity,
+    and the Activity's Retry Policy owns the retries.
+    """
+
+    def __init__(self, providers: Iterable[OpenAICompatibleProvider]) -> None:
+        self._providers = {provider.name: provider for provider in providers}
+        self._clients: dict[str, AsyncOpenAI] = {}
+        for provider in self._providers.values():
+            api_key = os.environ.get(provider.api_key_env)
+            if api_key:
+                self._clients[provider.name] = AsyncOpenAI(
+                    base_url=provider.base_url, api_key=api_key, max_retries=0
+                )
+
+    @property
+    def configured(self) -> list[str]:
+        """Names of the providers whose API key is set."""
+        return sorted(self._clients)
+
+    def get_model(self, model_name: str | None) -> Model:
+        provider_name, _, model = (model_name or "").partition("/")
+        provider = self._providers.get(provider_name)
+        if provider is None or not model:
+            # A mistake in the Workflow's input: fail the Activity without retrying.
+            raise ApplicationError(
+                f"Model name {model_name!r} must be <provider>/<model> with a "
+                f"provider in {sorted(self._providers)}",
+                non_retryable=True,
+            )
+        client = self._clients.get(provider_name)
+        if client is None:
+            raise ApplicationError(
+                f"Set {provider.api_key_env} and restart the Worker to use {provider_name}",
+                non_retryable=True,
+            )
+        # Chat Completions is the API all of these providers serve. The Agents
+        # SDK defaults to the Responses API, which not every provider offers.
+        return OpenAIChatCompletionsModel(model=model, openai_client=client)
+
+
+# @@@SNIPEND

openai_agents/model_providers/run_openai_compatible_worker.py

@@ -0,0 +1,66 @@
+"""Worker that serves Baseten, Fireworks AI, and Together AI from one process."""
+
+import asyncio
+import logging
+import sys
+from datetime import timedelta
+
+from agents import set_tracing_disabled
+from temporalio.client import Client
+from temporalio.openai_agents import ModelActivityParameters, OpenAIAgentsPlugin
+from temporalio.worker import Worker
+
+from openai_agents.model_providers.openai_compatible_providers import (
+    PROVIDERS,
+    OpenAICompatibleModelProvider,
+)
+from openai_agents.model_providers.workflows.openai_compatible_workflow import (
+    OpenAICompatibleAgentWorkflow,
+)
+
+
+async def main() -> None:
+    # Disable Agents SDK tracing: the default exporter sends traces to OpenAI's
+    # backend, which needs an OpenAI API key that this sample does not have.
+    set_tracing_disabled(disabled=True)
+
+    logging.basicConfig(level=logging.WARNING)
+    logging.getLogger("temporalio.workflow").setLevel(logging.DEBUG)
+
+    # @@@SNIPSTART python-openai-agents-openai-compatible-worker
+    # One provider for every endpoint whose API key is set. The model name on
+    # each Workflow's agent selects the endpoint.
+    model_provider = OpenAICompatibleModelProvider(PROVIDERS.values())
+    if not model_provider.configured:
+        env_vars = ", ".join(provider.api_key_env for provider in PROVIDERS.values())
+        sys.exit(f"No provider API key is set; export at least one of {env_vars}.")
+    print(f"Serving providers: {', '.join(model_provider.configured)}", flush=True)
+
+    client = await Client.connect(
+        "localhost:7233",
+        plugins=[
+            OpenAIAgentsPlugin(
+                model_params=ModelActivityParameters(
+                    # A custom model provider must set a timeout. Raise it for
+                    # slow models or deployments that scale up from zero.
+                    start_to_close_timeout=timedelta(seconds=60),
+                    # The plugin heartbeats, so a Worker that dies mid-call is
+                    # noticed in seconds rather than after the full timeout.
+                    heartbeat_timeout=timedelta(seconds=10),
+                ),
+                model_provider=model_provider,
+            ),
+        ],
+    )
+
+    worker = Worker(
+        client,
+        task_queue="openai-agents-model-providers-task-queue",
+        workflows=[OpenAICompatibleAgentWorkflow],
+    )
+    await worker.run()
+    # @@@SNIPEND
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

openai_agents/model_providers/run_openai_compatible_workflow.py

@@ -0,0 +1,67 @@
+"""Start one Workflow per model name; they run concurrently on the same Worker."""
+
+import argparse
+import asyncio
+import os
+
+from temporalio.client import Client, WorkflowFailureError
+from temporalio.exceptions import FailureError
+from temporalio.openai_agents import OpenAIAgentsPlugin
+
+from openai_agents.model_providers.openai_compatible_providers import PROVIDERS
+from openai_agents.model_providers.workflows.openai_compatible_workflow import (
+    OpenAICompatibleAgentInput,
+    OpenAICompatibleAgentWorkflow,
+)
+
+
+async def run_one(client: Client, model: str) -> None:
+    try:
+        result = await client.execute_workflow(
+            OpenAICompatibleAgentWorkflow.run,
+            OpenAICompatibleAgentInput(
+                prompt="What's the weather in Tokyo?", model=model
+            ),
+            id=f"openai-agents-openai-compatible-{model.replace('/', '-')}",
+            task_queue="openai-agents-model-providers-task-queue",
+        )
+        print(f"{model}:\n{result}\n")
+    except WorkflowFailureError as e:
+        # Report the root cause, such as the provider's "Set FIREWORKS_API_KEY".
+        cause = e.cause
+        while isinstance(cause, FailureError) and cause.cause is not None:
+            cause = cause.cause
+        print(f"{model}: failed: {cause}\n")
+
+
+async def main(models: list[str]) -> None:
+    client = await Client.connect(
+        "localhost:7233",
+        plugins=[
+            OpenAIAgentsPlugin(),
+        ],
+    )
+    # Each Workflow names its own model, so these run side by side on one
+    # Worker, each on its own provider.
+    await asyncio.gather(*(run_one(client, model) for model in models))
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--model",
+        action="append",
+        dest="models",
+        metavar="PROVIDER/MODEL",
+        help="model to run, such as together/openai/gpt-oss-120b; repeat for "
+        "several. Defaults to each provider whose API key is set in this shell.",
+    )
+    args = parser.parse_args()
+    models = args.models or [
+        f"{provider.name}/{provider.default_model}"
+        for provider in PROVIDERS.values()
+        if os.environ.get(provider.api_key_env)
+    ]
+    if not models:
+        parser.error("pass --model or export a provider API key")
+    asyncio.run(main(models))

openai_agents/model_providers/workflows/openai_compatible_workflow.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from agents import Agent, Runner, function_tool
+from temporalio import workflow
+
+
+# @@@SNIPSTART python-openai-agents-openai-compatible-workflow
+@dataclass
+class OpenAICompatibleAgentInput:
+    prompt: str
+    # "<provider>/<model>", such as "together/openai/gpt-oss-120b". The Worker's
+    # model provider routes the agent's model calls to that provider.
+    model: str
+
+
+@workflow.defn
+class OpenAICompatibleAgentWorkflow:
+    @workflow.run
+    async def run(self, input: OpenAICompatibleAgentInput) -> str:
+        # Tools that run inside the Workflow must be async: the Agents SDK runs
+        # sync tools in a thread, which the Workflow sandbox does not allow.
+        @function_tool
+        async def get_weather(city: str) -> str:
+            workflow.logger.debug(f"Getting weather for {city}")
+            return f"The weather in {city} is sunny."
+
+        agent = Agent(
+            name="Assistant",
+            instructions="You only respond in haikus. When asked about the weather always use the tool to get the current weather.",
+            model=input.model,
+            tools=[get_weather],
+        )
+
+        result = await Runner.run(agent, input.prompt)
+        return result.final_output
+
+
+# @@@SNIPEND