Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -117,6 +117,7 @@
from ._shared import (
AzureTokenProvider,
_attach_prompt_cache_breakpoint, # pyright: ignore[reportPrivateUsage]
_is_mp3_media_type, # pyright: ignore[reportPrivateUsage]
load_openai_service_settings,
maybe_append_azure_endpoint_guidance,
)
Expand Down Expand Up @@ -2150,7 +2151,7 @@ def _prepare_content_for_openai(
if content.has_top_level_media_type("audio"):
if content.media_type and "wav" in content.media_type:
format = "wav"
elif content.media_type and "mp3" in content.media_type:
elif _is_mp3_media_type(content.media_type):
format = "mp3"
else:
logger.warning("Unsupported audio media type: %s", content.media_type)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@
PROMPT_CACHE_BREAKPOINT_KEY,
AzureTokenProvider,
_attach_prompt_cache_breakpoint, # pyright: ignore[reportPrivateUsage]
_is_mp3_media_type, # pyright: ignore[reportPrivateUsage]
load_openai_service_settings,
maybe_append_azure_endpoint_guidance,
)
Expand Down Expand Up @@ -1315,7 +1316,7 @@ def _prepare_content_for_openai(self, content: Content) -> dict[str, Any]:
case "data" | "uri" if content.has_top_level_media_type("audio"):
if content.media_type and "wav" in content.media_type:
audio_format = "wav"
elif content.media_type and "mp3" in content.media_type:
elif _is_mp3_media_type(content.media_type):
audio_format = "mp3"
else:
logger.debug("Unsupported audio media type: %s", content.media_type)
Expand Down
23 changes: 23 additions & 0 deletions python/packages/openai/agent_framework_openai/_shared.py
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,29 @@
PROMPT_CACHE_BREAKPOINT_KEY = "prompt_cache_breakpoint"


MP3_MEDIA_TYPES = frozenset({"audio/mp3", "audio/mpeg", "audio/x-mpeg", "audio/mpeg3", "audio/x-mpeg-3"})


def _is_mp3_media_type( # pyright: ignore[reportUnusedFunction]
media_type: str | None,
) -> bool:
"""Check whether an audio media type names MP3 audio.

The media type is stripped of parameters and case-folded before it is compared against
the registered MP3 aliases, so that sibling MPEG subtypes such as ``audio/mpegurl`` or
``audio/mpeg4-generic`` are not mistaken for MP3.

Args:
media_type: The media type of the audio content, if any.

Returns:
True if the media type is an alias of MP3 audio, False otherwise.
"""
if not media_type:
return False
return media_type.partition(";")[0].strip().lower() in MP3_MEDIA_TYPES


def _attach_prompt_cache_breakpoint( # pyright: ignore[reportUnusedFunction]
part: dict[str, Any], content: Content
) -> dict[str, Any]:
Expand Down
26 changes: 26 additions & 0 deletions python/packages/openai/tests/openai/test_openai_chat_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -7717,6 +7717,32 @@ def test_prepare_content_for_openai_audio_content() -> None:
assert result["type"] == "input_audio"
assert result["input_audio"]["format"] == "mp3"

# Test MP3 audio content with the registered media type
mpeg_content = Content.from_uri(uri="data:audio/mpeg;base64,ghi789", media_type="audio/mpeg")
result = client._prepare_content_for_openai("user", mpeg_content)
assert result["type"] == "input_audio"
assert result["input_audio"]["data"] == "data:audio/mpeg;base64,ghi789"
assert result["input_audio"]["format"] == "mp3"

# Test MP3 aliases with media type parameters and mixed case
parameterized_content = Content.from_uri(
uri="data:audio/mpeg;base64,jkl012",
media_type="Audio/MPEG; codecs=mpeg-3",
)
result = client._prepare_content_for_openai("user", parameterized_content)
assert result["type"] == "input_audio"
assert result["input_audio"]["format"] == "mp3"


def test_prepare_content_for_openai_non_mp3_mpeg_audio_is_unsupported() -> None:
"""Test _prepare_content_for_openai omits MPEG audio that is not MP3."""
client = OpenAIChatClient(model="test-model", api_key="test-key")

for media_type in ("audio/mpegurl", "audio/mpeg4-generic"):
content = Content.from_uri(uri=f"data:{media_type};base64,abc123", media_type=media_type)

assert client._prepare_content_for_openai("user", content) == {}


def test_prepare_content_for_openai_unsupported_content() -> None:
"""Test _prepare_content_for_openai with unsupported content types."""
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -602,11 +602,44 @@ def test_prepare_content_for_openai_data_content_image(
assert result["input_audio"]["data"] == "//uQAAAAWGluZwAAAA8AAAACAAACcQ=="
assert result["input_audio"]["format"] == "mp3"

# Test DataContent with MP3 audio using the registered media type
mpeg_data_content = Content.from_uri(
uri="data:audio/mpeg;base64,//uQAAAAWGluZwAAAA8AAAACAAACcQ==",
media_type="audio/mpeg",
)

result = client._prepare_content_for_openai(mpeg_data_content) # type: ignore

assert result["type"] == "input_audio"
assert result["input_audio"]["data"] == "//uQAAAAWGluZwAAAA8AAAACAAACcQ=="
assert result["input_audio"]["format"] == "mp3"

# Test MP3 aliases with media type parameters and mixed case
parameterized_content = Content.from_uri(
uri="data:audio/mpeg;base64,//uQAAAAWGluZwAAAA8AAAACAAACcQ==",
media_type="Audio/MPEG; codecs=mpeg-3",
)

result = client._prepare_content_for_openai(parameterized_content) # type: ignore

assert result["type"] == "input_audio"
assert result["input_audio"]["format"] == "mp3"

unsupported_audio = Content.from_uri(uri="data:audio/ogg;base64,abc123", media_type="audio/ogg")

assert client._prepare_content_for_openai(unsupported_audio) == {} # type: ignore


def test_prepare_content_for_openai_non_mp3_mpeg_audio_is_unsupported() -> None:
"""Test _prepare_content_for_openai omits MPEG audio that is not MP3."""
client = OpenAIChatCompletionClient(model="test-model", api_key="test-key")

for media_type in ("audio/mpegurl", "audio/mpeg4-generic"):
content = Content.from_uri(uri=f"data:{media_type};base64,abc123", media_type=media_type)

assert client._prepare_content_for_openai(content) == {} # type: ignore


def test_prepare_message_for_openai_omits_unsupported_content() -> None:
client = OpenAIChatCompletionClient(model="test-model", api_key="test-key")
unsupported = Content.from_uri(uri="data:text/plain;base64,SGVsbG8=", media_type="text/plain")
Expand Down
Loading