wip

Author: rm-openai

src/agents/agent.py

@@ -94,6 +94,33 @@ class AgentBase:
     mcp_config: MCPConfig = field(default_factory=lambda: MCPConfig())
     """Configuration for MCP servers."""
 
+    async def get_mcp_tools(self, run_context: RunContextWrapper[TContext]) -> list[Tool]:
+        """Fetches the available tools from the MCP servers."""
+        convert_schemas_to_strict = self.mcp_config.get("convert_schemas_to_strict", False)
+        return await MCPUtil.get_all_function_tools(
+            self.mcp_servers, convert_schemas_to_strict, run_context, self
+        )
+
+    async def get_all_tools(self, run_context: RunContextWrapper[Any]) -> list[Tool]:
+        """All agent tools, including MCP tools and function tools."""
+        mcp_tools = await self.get_mcp_tools(run_context)
+
+        async def _check_tool_enabled(tool: Tool) -> bool:
+            if not isinstance(tool, FunctionTool):
+                return True
+
+            attr = tool.is_enabled
+            if isinstance(attr, bool):
+                return attr
+            res = attr(run_context, self)
+            if inspect.isawaitable(res):
+                return bool(await res)
+            return bool(res)
+
+        results = await asyncio.gather(*(_check_tool_enabled(t) for t in self.tools))
+        enabled: list[Tool] = [t for t, ok in zip(self.tools, results) if ok]
+        return [*mcp_tools, *enabled]
+
 
 @dataclass
 class Agent(AgentBase, Generic[TContext]):
@@ -262,30 +289,3 @@ async def get_prompt(
     ) -> ResponsePromptParam | None:
         """Get the prompt for the agent."""
         return await PromptUtil.to_model_input(self.prompt, run_context, self)
-
-    async def get_mcp_tools(self, run_context: RunContextWrapper[TContext]) -> list[Tool]:
-        """Fetches the available tools from the MCP servers."""
-        convert_schemas_to_strict = self.mcp_config.get("convert_schemas_to_strict", False)
-        return await MCPUtil.get_all_function_tools(
-            self.mcp_servers, convert_schemas_to_strict, run_context, self
-        )
-
-    async def get_all_tools(self, run_context: RunContextWrapper[Any]) -> list[Tool]:
-        """All agent tools, including MCP tools and function tools."""
-        mcp_tools = await self.get_mcp_tools(run_context)
-
-        async def _check_tool_enabled(tool: Tool) -> bool:
-            if not isinstance(tool, FunctionTool):
-                return True
-
-            attr = tool.is_enabled
-            if isinstance(attr, bool):
-                return attr
-            res = attr(run_context, self)
-            if inspect.isawaitable(res):
-                return bool(await res)
-            return bool(res)
-
-        results = await asyncio.gather(*(_check_tool_enabled(t) for t in self.tools))
-        enabled: list[Tool] = [t for t, ok in zip(self.tools, results) if ok]
-        return [*mcp_tools, *enabled]

src/agents/model_settings.py

@@ -17,9 +17,9 @@
 class _OmitTypeAnnotation:
     @classmethod
     def __get_pydantic_core_schema__(
-            cls,
-            _source_type: Any,
-            _handler: GetCoreSchemaHandler,
+        cls,
+        _source_type: Any,
+        _handler: GetCoreSchemaHandler,
     ) -> core_schema.CoreSchema:
         def validate_from_none(value: None) -> _Omit:
             return _Omit()
@@ -39,12 +39,14 @@ def validate_from_none(value: None) -> _Omit:
                     from_none_schema,
                 ]
             ),
-            serialization=core_schema.plain_serializer_function_ser_schema(
-                lambda instance: None
-            ),
+            serialization=core_schema.plain_serializer_function_ser_schema(lambda instance: None),
         )
+
+
 Omit = Annotated[_Omit, _OmitTypeAnnotation]
 Headers: TypeAlias = Mapping[str, Union[str, Omit]]
+ToolChoice: TypeAlias = Literal["auto", "required", "none"] | str | None
+
 
 @dataclass
 class ModelSettings:

src/agents/realtime/__init__.py

@@ -1,3 +1,51 @@
 from .agent import RealtimeAgent, RealtimeAgentHooks, RealtimeRunHooks
+from .config import APIKeyOrKeyFunc
+from .events import (
+    RealtimeAgentEndEvent,
+    RealtimeAgentStartEvent,
+    RealtimeAudio,
+    RealtimeAudioEnd,
+    RealtimeAudioInterrupted,
+    RealtimeError,
+    RealtimeHandoffEvent,
+    RealtimeHistoryAdded,
+    RealtimeHistoryUpdated,
+    RealtimeRawTransportEvent,
+    RealtimeSessionEvent,
+    RealtimeToolEnd,
+    RealtimeToolStart,
+)
+from .session import RealtimeSession
+from .transport import (
+    RealtimeModelName,
+    RealtimeSessionTransport,
+    RealtimeTransportConnectionOptions,
+    RealtimeTransportListener,
+)
 
-__all__ = ["RealtimeAgent", "RealtimeAgentHooks", "RealtimeRunHooks"]
+__all__ = [
+    "RealtimeAgent",
+    "RealtimeAgentHooks",
+    "RealtimeRunHooks",
+    "RealtimeSession",
+    "RealtimeSessionListener",
+    "RealtimeSessionListenerFunc",
+    "APIKeyOrKeyFunc",
+    "RealtimeModelName",
+    "RealtimeSessionTransport",
+    "RealtimeTransportListener",
+    "RealtimeTransportConnectionOptions",
+    "RealtimeSessionEvent",
+    "RealtimeAgentStartEvent",
+    "RealtimeAgentEndEvent",
+    "RealtimeHandoffEvent",
+    "RealtimeToolStart",
+    "RealtimeToolEnd",
+    "RealtimeRawTransportEvent",
+    "RealtimeAudioEnd",
+    "RealtimeAudio",
+    "RealtimeAudioInterrupted",
+    "RealtimeError",
+    "RealtimeHistoryUpdated",
+    "RealtimeHistoryAdded",
+]

src/agents/realtime/agent.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import asyncio
 import dataclasses
 import inspect
 from collections.abc import Awaitable
@@ -10,9 +9,7 @@
 from ..agent import AgentBase
 from ..lifecycle import AgentHooksBase, RunHooksBase
 from ..logger import logger
-from ..mcp import MCPUtil
 from ..run_context import RunContextWrapper, TContext
-from ..tool import FunctionTool, Tool
 from ..util._types import MaybeAwaitable
 
 RealtimeAgentHooks = AgentHooksBase[TContext, "RealtimeAgent[TContext]"]
@@ -81,30 +78,3 @@ async def get_system_prompt(self, run_context: RunContextWrapper[TContext]) -> s
             logger.error(f"Instructions must be a string or a function, got {self.instructions}")
 
         return None
-
-    async def get_mcp_tools(self, run_context: RunContextWrapper[TContext]) -> list[Tool]:
-        """Fetches the available tools from the MCP servers."""
-        convert_schemas_to_strict = self.mcp_config.get("convert_schemas_to_strict", False)
-        return await MCPUtil.get_all_function_tools(
-            self.mcp_servers, convert_schemas_to_strict, run_context, self
-        )
-
-    async def get_all_tools(self, run_context: RunContextWrapper[Any]) -> list[Tool]:
-        """All agent tools, including MCP tools and function tools."""
-        mcp_tools = await self.get_mcp_tools(run_context)
-
-        async def _check_tool_enabled(tool: Tool) -> bool:
-            if not isinstance(tool, FunctionTool):
-                return True
-
-            attr = tool.is_enabled
-            if isinstance(attr, bool):
-                return attr
-            res = attr(run_context, self)
-            if inspect.isawaitable(res):
-                return bool(await res)
-            return bool(res)
-
-        results = await asyncio.gather(*(_check_tool_enabled(t) for t in self.tools))
-        enabled: list[Tool] = [t for t, ok in zip(self.tools, results) if ok]
-        return [*mcp_tools, *enabled]

src/agents/realtime/config.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import inspect
+from typing import (
+    Callable,
+    Literal,
+    Union,
+)
+
+from typing_extensions import NotRequired, TypeAlias, TypedDict
+
+from ..model_settings import ToolChoice
+from ..tool import FunctionTool
+from ..util._types import MaybeAwaitable
+
+
+class RealtimeClientMessage(TypedDict, total=False):
+    type: str  # explicitly required
+    # All additional keys are permitted because total=False
+
+
+class UserInputText(TypedDict):
+    type: Literal["input_text"]
+    text: str
+
+
+class RealtimeUserInputMessage(TypedDict):
+    type: Literal["message"]
+    role: Literal["user"]
+    content: list[UserInputText]
+
+
+RealtimeUserInput: TypeAlias = Union[str, RealtimeUserInputMessage]
+
+
+RealtimeAudioFormat: TypeAlias = Union[Literal["pcm16", "g711_ulaw", "g711_alaw"], str]
+
+
+class RealtimeInputAudioTranscriptionConfig(TypedDict, total=False):
+    language: NotRequired[str]
+    model: NotRequired[Literal["gpt-4o-transcribe", "gpt-4o-mini-transcribe", "whisper-1"] | str]
+    prompt: NotRequired[str]
+
+
+class RealtimeTurnDetectionConfig(TypedDict, total=False):
+    """Turn detection config. Allows extra vendor keys if needed."""
+
+    type: NotRequired[Literal["semantic_vad", "server_vad"]]
+    create_response: NotRequired[bool]
+    eagerness: NotRequired[Literal["auto", "low", "medium", "high"]]
+    interrupt_response: NotRequired[bool]
+    prefix_padding_ms: NotRequired[int]
+    silence_duration_ms: NotRequired[int]
+    threshold: NotRequired[float]
+
+
+class RealtimeSessionConfig(TypedDict):
+    api_key: NotRequired[APIKeyOrKeyFunc]
+    model: NotRequired[str]
+    instructions: NotRequired[str]
+    modalities: NotRequired[list[Literal["text", "audio"]]]
+    voice: NotRequired[str]
+
+    input_audio_format: NotRequired[RealtimeAudioFormat]
+    output_audio_format: NotRequired[RealtimeAudioFormat]
+    input_audio_transcription: NotRequired[RealtimeInputAudioTranscriptionConfig]
+    turn_detection: NotRequired[RealtimeTurnDetectionConfig]
+
+    tool_choice: NotRequired[ToolChoice]
+    tools: NotRequired[list[FunctionTool]]
+
+
+APIKeyOrKeyFunc = str | Callable[[], MaybeAwaitable[str]]
+"""Either an API key or a function that returns an API key."""
+
+
+async def get_api_key(key: APIKeyOrKeyFunc | None) -> str | None:
+    """Get the API key from the key or key function."""
+    if key is None:
+        return None
+    elif isinstance(key, str):
+        return key
+
+    result = key()
+    if inspect.isawaitable(result):
+        return await result
+    return result
+
+    # TODO (rm) Add tracing support
+    # tracing: NotRequired[RealtimeTracingConfig | None]