Python: follow on work on OpenAI (#169)

* updated openai, fcc works, with sample

* reduced files in openai

python/packages/main/agent_framework/__init__.py

@@ -19,6 +19,7 @@ _IMPORTS = {
     "AgentThread": "._agents",
     "AITool": "._tools",
     "ai_function": "._tools",
+    "AIFunction": "._tools",
     "AIContent": "._types",
     "AIContents": "._types",
     "ChatClientAgent": "._agents",

python/packages/main/agent_framework/__init__.pyi

@@ -5,7 +5,7 @@ from ._agents import Agent, AgentThread, ChatClientAgent, ChatClientAgentThread,
 from ._clients import ChatClient, ChatClientBase, EmbeddingGenerator, use_tool_calling
 from ._logging import get_logger
 from ._pydantic import AFBaseModel, AFBaseSettings
-from ._tools import AITool, ai_function
+from ._tools import AIFunction, AITool, ai_function
 from ._types import (
     AgentRunResponse,
     AgentRunResponseUpdate,
@@ -39,6 +39,7 @@ __all__ = [
     "AFBaseSettings",
     "AIContent",
     "AIContents",
+    "AIFunction",
     "AITool",
     "Agent",
     "AgentRunResponse",

python/packages/main/agent_framework/_clients.py

@@ -108,7 +108,10 @@ def _prepare_tools_and_tool_choice(chat_options: ChatOptions) -> None:
     chat_options.tools = [
         (_tool_to_json_schema_spec(t) if isinstance(t, AITool) else t) for t in chat_options.tools or []
     ]
-    chat_options.tool_choice = chat_tool_mode.mode
+    if not chat_options.tools:
+        chat_options.tool_choice = ChatToolMode.NONE.mode
+    else:
+        chat_options.tool_choice = chat_tool_mode.mode
 
 
 def _tool_call_non_streaming(func: TInnerGetResponse) -> TInnerGetResponse:
@@ -208,8 +211,9 @@ def _tool_call_streaming(func: TInnerGetStreamingResponse) -> TInnerGetStreaming
             # the full completion depending on the prompt, the message may contain both function call
             # content and others
             response: ChatResponse = ChatResponse.from_chat_response_updates(all_messages)
-            function_calls = [item for item in response.messages[0].contents if isinstance(item, FunctionCallContent)]
+            # add the single assistant response message to the history
             messages.append(response.messages[0])
+            function_calls = [item for item in response.messages[0].contents if isinstance(item, FunctionCallContent)]
 
             if function_calls:
                 # Run all function calls concurrently
@@ -224,8 +228,9 @@ def _tool_call_streaming(func: TInnerGetStreamingResponse) -> TInnerGetStreaming
                     for seq_idx, function_call in enumerate(function_calls)
                 ])
                 yield ChatResponseUpdate(contents=results, role="tool")
-                response.messages.append(ChatMessage(role="tool", contents=results))
-                messages.extend(response.messages)
+                function_result_msg = ChatMessage(role="tool", contents=results)
+                response.messages.append(function_result_msg)
+                messages.append(function_result_msg)
                 continue
 
         # Failsafe: give up on tools, ask model for plain answer
@@ -238,11 +243,17 @@ def _tool_call_streaming(func: TInnerGetStreamingResponse) -> TInnerGetStreaming
 
 
 def use_tool_calling(cls: type[TChatClientBase]) -> type[TChatClientBase]:
-    inner_response = getattr(cls, "_inner_get_response", None)
-    if inner_response is not None:
+    """Class decorator that enables tool calling for a chat client.
+
+    Remarks:
+        This only works on classes that derive from ChatClientBase
+        and have the _tool_map attribute as well as the _inner_get_response
+        and _inner_get_streaming_response methods.
+
+    """
+    if inner_response := getattr(cls, "_inner_get_response", None):
         cls._inner_get_response = _tool_call_non_streaming(inner_response)  # type: ignore
-    inner_streaming_response = getattr(cls, "_inner_get_streaming_response", None)
-    if inner_streaming_response is not None:
+    if inner_streaming_response := getattr(cls, "_inner_get_streaming_response", None):
         cls._inner_get_streaming_response = _tool_call_streaming(inner_streaming_response)  # type: ignore
     return cls
 
@@ -303,6 +314,17 @@ class ChatClientBase(AFBaseModel, ABC):
     maximum_iterations_per_request: int = 10
     _tool_map: dict[str, AIFunction[BaseModel, Any]] = PrivateAttr(default_factory=dict)  # type: ignore
 
+    def _prepare_messages(self, messages: str | ChatMessage | list[str | ChatMessage]) -> MutableSequence[ChatMessage]:
+        """Turn the allowed input into a list of chat messages."""
+        if isinstance(messages, str):
+            messages = [ChatMessage(role="user", text=messages)]
+        if isinstance(messages, ChatMessage):
+            messages = [messages]
+        for i, msg in enumerate(messages):
+            if isinstance(msg, str):
+                messages[i] = ChatMessage(role="user", text=msg)
+        return messages  # type: ignore[return-value]
+
     # region Internal methods to be implemented by the derived classes
 
     @abstractmethod
@@ -355,14 +377,14 @@ class ChatClientBase(AFBaseModel, ABC):
 
     async def get_response(
         self,
-        messages: str | ChatMessage | list[ChatMessage],
+        messages: str | ChatMessage | list[str | ChatMessage],
         *,
         model: str | None = None,
         max_tokens: int | None = None,
         temperature: float | None = None,
         top_p: float | None = None,
-        tool_choice: ChatToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = None,
-        tools: Sequence[AITool] | None = None,
+        tool_choice: ChatToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = "auto",
+        tools: AITool | Sequence[AITool] | None = None,
         response_format: type[BaseModel] | None = None,
         user: str | None = None,
         stop: str | Sequence[str] | None = None,
@@ -402,6 +424,8 @@ class ChatClientBase(AFBaseModel, ABC):
             A chat response from the model.
         """
         if tools is not None:
+            if not isinstance(tools, Sequence):
+                tools = [tools]
             self._tool_map = {tool.name: tool for tool in tools if isinstance(tool, AIFunction)}
         chat_options = ChatOptions(
             ai_model_id=model,
@@ -421,23 +445,20 @@ class ChatClientBase(AFBaseModel, ABC):
             metadata=metadata,
             additional_properties=additional_properties or {},
         )
-        if isinstance(messages, str):
-            messages = [ChatMessage(role="user", text=messages)]
-        if isinstance(messages, ChatMessage):
-            messages = [messages]
+        prepped_messages = self._prepare_messages(messages)
         _prepare_tools_and_tool_choice(chat_options=chat_options)
-        return await self._inner_get_response(messages=messages, chat_options=chat_options, **kwargs)
+        return await self._inner_get_response(messages=prepped_messages, chat_options=chat_options, **kwargs)
 
     async def get_streaming_response(
         self,
-        messages: str | ChatMessage | list[ChatMessage],
+        messages: str | ChatMessage | list[str | ChatMessage],
         *,
         model: str | None = None,
         max_tokens: int | None = None,
         temperature: float | None = None,
         top_p: float | None = None,
-        tool_choice: ChatToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = None,
-        tools: Sequence[AITool] | None = None,
+        tool_choice: ChatToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = "auto",
+        tools: AITool | Sequence[AITool] | None = None,
         response_format: type[BaseModel] | None = None,
         user: str | None = None,
         stop: str | Sequence[str] | None = None,
@@ -476,6 +497,8 @@ class ChatClientBase(AFBaseModel, ABC):
             A stream representing the response(s) from the LLM.
         """
         if tools is not None:
+            if not isinstance(tools, Sequence):
+                tools = [tools]
             self._tool_map = {tool.name: tool for tool in tools if isinstance(tool, AIFunction)}
         chat_options = ChatOptions(
             ai_model_id=model,
@@ -496,12 +519,11 @@ class ChatClientBase(AFBaseModel, ABC):
             additional_properties=additional_properties or {},
             **kwargs,
         )
-        if isinstance(messages, str):
-            messages = [ChatMessage(role="user", text=messages)]
-        if isinstance(messages, ChatMessage):
-            messages = [messages]
+        prepped_messages = self._prepare_messages(messages)
         _prepare_tools_and_tool_choice(chat_options=chat_options)
-        async for update in self._inner_get_streaming_response(messages=messages, chat_options=chat_options, **kwargs):
+        async for update in self._inner_get_streaming_response(
+            messages=prepped_messages, chat_options=chat_options, **kwargs
+        ):
             yield update
 
 

python/packages/main/agent_framework/_tools.py

@@ -1,8 +1,8 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-import functools
 import inspect
 from collections.abc import Awaitable, Callable, Mapping
+from functools import wraps
 from typing import Any, Generic, Protocol, TypeVar, runtime_checkable
 
 from pydantic import BaseModel, create_model
@@ -10,7 +10,7 @@ from pydantic import BaseModel, create_model
 
 @runtime_checkable
 class AITool(Protocol):
-    """Represents a tool that can be specified to an AI service."""
+    """Represents a generic tool that can be specified to an AI service."""
 
     name: str
     """The name of the tool."""
@@ -33,7 +33,7 @@ ReturnT = TypeVar("ReturnT")
 
 
 class AIFunction(AITool, Generic[ArgsT, ReturnT]):
-    """A tool that represents a function that can be called by an AI service."""
+    """A AITool that is callable as code."""
 
     def __init__(
         self,
@@ -98,8 +98,18 @@ def ai_function(
     name: str | None = None,
     description: str | None = None,
     additional_properties: dict[str, Any] | None = None,
-) -> AIFunction[Any, ReturnT] | Callable[[Callable[..., ReturnT | Awaitable[ReturnT]]], AIFunction[Any, ReturnT]]:
-    """Decorate a function to turn it into a AIFunction that can be passed to models.
+) -> AIFunction[Any, ReturnT]:
+    """Decorate a function to turn it into a AIFunction that can be passed to models and executed automatically.
+
+    Remarks:
+        In order to add descriptions to parameters, use:
+
+        ```python
+        from typing import Annotated
+        from pydantic import Field
+
+        arg: Annotated[<type>, Field(description="<description>")]
+        ```
 
     Args:
         func: The function to wrap. If None, returns a decorator.
@@ -109,31 +119,32 @@ def ai_function(
 
     """
 
-    def wrapper(f: Callable[..., ReturnT | Awaitable[ReturnT]]) -> AIFunction[Any, ReturnT]:
-        tool_name: str = name or getattr(f, "__name__", "unknown_function")  # type: ignore[assignment]
-        tool_desc: str = description or (f.__doc__ or "")
-        sig = inspect.signature(f)
-        fields = {
-            pname: (
-                param.annotation if param.annotation is not inspect.Parameter.empty else str,
-                param.default if param.default is not inspect.Parameter.empty else ...,
-            )
-            for pname, param in sig.parameters.items()
-            if pname not in {"self", "cls"}
-        }
-        input_model: Any = create_model(f"{tool_name}_input", **fields)  # type: ignore[call-overload]
-        if not issubclass(input_model, BaseModel):
-            raise TypeError(f"Input model for {tool_name} must be a subclass of BaseModel, got {input_model}")
+    def decorator(func: Callable[..., ReturnT | Awaitable[ReturnT]]) -> AIFunction[Any, ReturnT]:
+        @wraps(func)
+        def wrapper(f: Callable[..., ReturnT | Awaitable[ReturnT]]) -> AIFunction[Any, ReturnT]:
+            tool_name: str = name or getattr(f, "__name__", "unknown_function")  # type: ignore[assignment]
+            tool_desc: str = description or (f.__doc__ or "")
+            sig = inspect.signature(f)
+            fields = {
+                pname: (
+                    param.annotation if param.annotation is not inspect.Parameter.empty else str,
+                    param.default if param.default is not inspect.Parameter.empty else ...,
+                )
+                for pname, param in sig.parameters.items()
+                if pname not in {"self", "cls"}
+            }
+            input_model: Any = create_model(f"{tool_name}_input", **fields)  # type: ignore[call-overload]
+            if not issubclass(input_model, BaseModel):
+                raise TypeError(f"Input model for {tool_name} must be a subclass of BaseModel, got {input_model}")
 
-        return functools.update_wrapper(  # type: ignore[return-value]
-            AIFunction[Any, ReturnT](
+            return AIFunction[Any, ReturnT](
                 func=f,
                 name=tool_name,
                 description=tool_desc,
                 input_model=input_model,
                 **(additional_properties if additional_properties is not None else {}),
-            ),
-            f,
-        )
+            )
 
-    return wrapper(func) if func else wrapper
+        return wrapper(func)
+
+    return decorator(func) if func else decorator  # type: ignore[reportReturnType, return-value]

python/packages/main/agent_framework/_types.py

@@ -4,15 +4,14 @@ import base64
 import json
 import re
 import sys
-from collections.abc import AsyncIterable, Iterable, Iterator, Mapping, MutableSequence, Sequence
+from collections.abc import AsyncIterable, Iterable, Iterator, Mapping, MutableMapping, MutableSequence, Sequence
 from typing import Annotated, Any, ClassVar, Generic, Literal, TypeVar, overload
 
 from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator
 
-from agent_framework.exceptions import AgentFrameworkException
-
 from ._pydantic import AFBaseModel
 from ._tools import AITool
+from .exceptions import AgentFrameworkException
 
 if sys.version_info >= (3, 11):
     from typing import Self  # pragma: no cover
@@ -224,7 +223,7 @@ def _coalesce_text_content(
                 first_new_content = i
         else:
             if first_new_content is not None:
-                new_content = type_(text="\n".join(current_texts))
+                new_content = type_(text=" ".join(current_texts))
                 new_content.raw_representation = contents[first_new_content].raw_representation
                 new_content.additional_properties = contents[first_new_content].additional_properties
                 # Store the replacement node. We inherit the properties of the first text node. We don't
@@ -235,7 +234,7 @@ def _coalesce_text_content(
                 first_new_content = None
             coalesced_contents.append(content)
     if current_texts:
-        coalesced_contents.append(type_(text="\n".join(current_texts)))
+        coalesced_contents.append(type_(text=" ".join(current_texts)))
     contents.clear()
     contents.extend(coalesced_contents)
 
@@ -647,7 +646,7 @@ class FunctionCallContent(AIContent):
     def __add__(self, other: "FunctionCallContent") -> "FunctionCallContent":
         if not isinstance(other, FunctionCallContent):
             raise TypeError("Incompatible type")
-        if self.call_id != other.call_id:
+        if other.call_id and self.call_id != other.call_id:
             raise AgentFrameworkException("Incompatible function call contents")
         if not self.arguments:
             arguments = other.arguments
@@ -880,15 +879,6 @@ class ChatMessage(AFBaseModel):
     raw_representation: Any | None = None
     """The raw representation of the chat message from an underlying implementation."""
 
-    @property
-    def text(self) -> str:
-        """Returns the text content of the message.
-
-        Remarks:
-            This property concatenates the text of all TextContent objects in Contents.
-        """
-        return "\n".join(content.text for content in self.contents if isinstance(content, TextContent))
-
     @overload
     def __init__(
         self,
@@ -916,7 +906,7 @@ class ChatMessage(AFBaseModel):
         self,
         role: ChatRole | Literal["system", "user", "assistant", "tool"],
         *,
-        contents: list[AIContents],
+        contents: MutableSequence[AIContents],
         author_name: str | None = None,
         message_id: str | None = None,
         additional_properties: dict[str, Any] | None = None,
@@ -938,7 +928,7 @@ class ChatMessage(AFBaseModel):
         role: ChatRole | Literal["system", "user", "assistant", "tool"],
         *,
         text: str | None = None,
-        contents: list[AIContents] | None = None,
+        contents: MutableSequence[AIContents] | None = None,
         author_name: str | None = None,
         message_id: str | None = None,
         additional_properties: dict[str, Any] | None = None,
@@ -959,6 +949,15 @@ class ChatMessage(AFBaseModel):
             raw_representation=raw_representation,  # type: ignore[reportCallIssue]
         )
 
+    @property
+    def text(self) -> str:
+        """Returns the text content of the message.
+
+        Remarks:
+            This property concatenates the text of all TextContent objects in Contents.
+        """
+        return " ".join(content.text for content in self.contents if isinstance(content, TextContent))
+
 
 # region: ChatResponse
 
@@ -1123,7 +1122,10 @@ class ChatResponse(AFBaseModel):
     @property
     def text(self) -> str:
         """Returns the concatenated text of all messages in the response."""
-        return "\n".join(message.text for message in self.messages if isinstance(message, ChatMessage))
+        return ("\n".join(message.text for message in self.messages if isinstance(message, ChatMessage))).strip()
+
+    def __str__(self) -> str:
+        return self.text
 
 
 class StructuredResponse(ChatResponse, Generic[TValue]):
@@ -1343,7 +1345,10 @@ class ChatResponseUpdate(AFBaseModel):
     @property
     def text(self) -> str:
         """Returns the concatenated text of all contents in the update."""
-        return "\n".join(content.text for content in self.contents if isinstance(content, TextContent))
+        return "".join(content.text for content in self.contents if isinstance(content, TextContent))
+
+    def __str__(self) -> str:
+        return self.text
 
     def with_(self, contents: list[AIContent] | None = None, message_id: str | None = None) -> Self:
         """Returns a new instance with the specified contents and message_id."""
@@ -1398,19 +1403,19 @@ class ChatOptions(AFBaseModel):
     temperature: Annotated[float | None, Field(ge=0.0, le=2.0)] = None
     top_p: Annotated[float | None, Field(ge=0.0, le=1.0)] = None
     tool_choice: ChatToolMode | Literal["auto", "required", "none"] | Mapping[str, Any] | None = None
-    tools: Sequence[AITool] | Sequence[Mapping[str, Any]] | None = None
+    tools: Sequence[AITool] | Sequence[MutableMapping[str, Any]] | None = None
     response_format: type[BaseModel] | None = Field(
         default=None, description="Structured output response format schema. Must be a valid Pydantic model."
     )
     user: str | None = None
     stop: str | Sequence[str] | None = None
     frequency_penalty: Annotated[float | None, Field(ge=-2.0, le=2.0)] = None
-    logit_bias: Mapping[str | int, float] | None = None
+    logit_bias: MutableMapping[str | int, float] | None = None
     presence_penalty: Annotated[float | None, Field(ge=-2.0, le=2.0)] = None
     seed: int | None = None
     store: bool | None = None
-    metadata: Mapping[str, str] | None = None
-    additional_properties: Mapping[str, Any] = Field(
+    metadata: MutableMapping[str, str] | None = None
+    additional_properties: MutableMapping[str, Any] = Field(
         default_factory=dict, description="Provider-specific additional properties."
     )
 
@@ -1745,3 +1750,9 @@ class TextToSpeechOptions(AFBaseModel):
         for key in merged_exclude:
             settings.pop(key, None)
         return settings
+
+
+# endregion
+
+
+# endregion

python/packages/main/agent_framework/const.py

@@ -1,5 +0,0 @@
-# Copyright (c) Microsoft. All rights reserved.
-
-from typing import Final
-
-USER_AGENT: Final[str] = "User-Agent"

python/packages/main/agent_framework/openai/__init__.py

@@ -0,0 +1,10 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+
+from ._chat_client import OpenAIChatClient
+from ._shared import OpenAISettings
+
+__all__ = [
+    "OpenAIChatClient",
+    "OpenAISettings",
+]

python/packages/main/agent_framework/openai/_chat_client.py

@@ -0,0 +1,345 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import json
+from collections.abc import AsyncIterable, Mapping, MutableSequence, Sequence
+from datetime import datetime
+from itertools import chain
+from typing import Any, ClassVar, cast
+
+from openai import AsyncOpenAI, AsyncStream
+from openai.types import CompletionUsage
+from openai.types.chat.chat_completion import ChatCompletion, Choice
+from openai.types.chat.chat_completion_chunk import ChatCompletionChunk, ChoiceDeltaToolCall
+from openai.types.chat.chat_completion_chunk import Choice as ChunkChoice
+from openai.types.chat.chat_completion_message_tool_call import ChatCompletionMessageToolCall
+from pydantic import SecretStr, ValidationError
+
+from .._clients import ChatClientBase, use_tool_calling
+from .._types import (
+    AIContents,
+    ChatFinishReason,
+    ChatMessage,
+    ChatOptions,
+    ChatResponse,
+    ChatResponseUpdate,
+    ChatRole,
+    FunctionCallContent,
+    FunctionResultContent,
+    TextContent,
+    UsageDetails,
+)
+from ..exceptions import ServiceInitializationError, ServiceInvalidResponseError
+from ._shared import OpenAIConfigBase, OpenAIHandler, OpenAIModelTypes, OpenAISettings
+
+
+# Implements agent_framework.ChatClient protocol, through ChatClientBase
+@use_tool_calling
+class OpenAIChatClientBase(OpenAIHandler, ChatClientBase):
+    """OpenAI Chat completion class."""
+
+    MODEL_PROVIDER_NAME: ClassVar[str] = "openai"
+    SUPPORTS_FUNCTION_CALLING: ClassVar[bool] = True
+
+    # region Overriding base class methods
+    # most of the methods are overridden from the ChatCompletionClientBase class, otherwise it is mentioned
+
+    async def _inner_get_response(
+        self,
+        *,
+        messages: MutableSequence[ChatMessage],
+        chat_options: ChatOptions,
+        **kwargs: Any,
+    ) -> ChatResponse:
+        chat_options.additional_properties["stream"] = False
+        if not chat_options.ai_model_id:
+            chat_options.ai_model_id = self.ai_model_id
+
+        response = await self._send_request(chat_options, messages=self._prepare_chat_history_for_request(messages))
+        assert isinstance(response, ChatCompletion)  # nosec  # noqa: S101
+        response_metadata = self._get_metadata_from_chat_response(response)
+        return next(
+            self._create_chat_message_content(response, choice, response_metadata) for choice in response.choices
+        )
+
+    # @trace_streaming_chat_completion(MODEL_PROVIDER_NAME)
+    async def _inner_get_streaming_response(
+        self,
+        *,
+        messages: MutableSequence[ChatMessage],
+        chat_options: ChatOptions,
+        **kwargs: Any,
+    ) -> AsyncIterable[ChatResponseUpdate]:
+        chat_options.additional_properties["stream"] = True
+        chat_options.additional_properties["stream_options"] = {"include_usage": True}
+        chat_options.ai_model_id = chat_options.ai_model_id or self.ai_model_id
+
+        response = await self._send_request(chat_options, messages=self._prepare_chat_history_for_request(messages))
+        if not isinstance(response, AsyncStream):
+            raise ServiceInvalidResponseError("Expected an AsyncStream[ChatCompletionChunk] response.")
+        async for chunk in response:
+            if len(chunk.choices) == 0 and chunk.usage is None:
+                continue
+
+            assert isinstance(chunk, ChatCompletionChunk)  # nosec  # noqa: S101
+            chunk_metadata = self._get_metadata_from_streaming_chat_response(chunk)
+            if chunk.usage is not None:
+                # Usage is contained in the last chunk where the choices are empty
+                # We are duplicating the usage metadata to all the choices in the response
+                yield ChatResponseUpdate(
+                    role=ChatRole.ASSISTANT,
+                    contents=[],
+                    ai_model_id=chat_options.ai_model_id,
+                    additional_properties=chunk_metadata,
+                )
+
+            else:
+                yield next(
+                    self._create_streaming_chat_message_content(chunk, choice, chunk_metadata)
+                    for choice in chunk.choices
+                )
+
+    # endregion
+
+    # region content creation
+
+    def _create_chat_message_content(
+        self, response: ChatCompletion, choice: Choice, response_metadata: dict[str, Any]
+    ) -> "ChatResponse":
+        """Create a chat message content object from a choice."""
+        metadata = self._get_metadata_from_chat_choice(choice)
+        metadata.update(response_metadata)
+        items: MutableSequence[ChatMessage] = []
+        if parsed_tool_calls := [tool for tool in self._get_tool_calls_from_chat_choice(choice)]:
+            items.append(ChatMessage(role="assistant", contents=parsed_tool_calls))
+        if choice.message.content:
+            items.append(ChatMessage(role="assistant", text=choice.message.content))
+        elif hasattr(choice.message, "refusal") and choice.message.refusal:
+            items.append(ChatMessage(role="assistant", text=choice.message.refusal))
+
+        return ChatResponse(
+            response_id=response.id,
+            created_at=datetime.fromtimestamp(response.created).strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
+            usage_details=self._usage_details_from_openai(response.usage) if response.usage else None,
+            messages=items,
+            model_id=self.ai_model_id,
+            additional_properties=metadata,
+            finish_reason=(ChatFinishReason(value=choice.finish_reason) if choice.finish_reason else None),
+        )
+
+    def _create_streaming_chat_message_content(
+        self,
+        chunk: ChatCompletionChunk,
+        choice: ChunkChoice,
+        chunk_metadata: dict[str, Any],
+    ) -> ChatResponseUpdate:
+        """Create a streaming chat message content object from a choice."""
+        metadata = self._get_metadata_from_chat_choice(choice)
+        metadata.update(chunk_metadata)
+
+        items: list[Any] = self._get_tool_calls_from_chat_choice(choice)
+        if choice.delta and choice.delta.content is not None:
+            items.append(TextContent(text=choice.delta.content))
+        return ChatResponseUpdate(
+            created_at=datetime.fromtimestamp(chunk.created).strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
+            contents=items,
+            role=ChatRole.ASSISTANT,
+            ai_model_id=self.ai_model_id,
+            additional_properties=metadata,
+            finish_reason=(ChatFinishReason(value=choice.finish_reason) if choice.finish_reason else None),
+        )
+
+    def _usage_details_from_openai(self, usage: CompletionUsage) -> UsageDetails | None:
+        return UsageDetails(
+            prompt_tokens=usage.prompt_tokens,
+            completion_tokens=usage.completion_tokens,
+            total_tokens=usage.total_tokens,
+        )
+
+    def _get_metadata_from_chat_response(self, response: ChatCompletion) -> dict[str, Any]:
+        """Get metadata from a chat response."""
+        return {
+            "system_fingerprint": response.system_fingerprint,
+        }
+
+    def _get_metadata_from_streaming_chat_response(self, response: ChatCompletionChunk) -> dict[str, Any]:
+        """Get metadata from a streaming chat response."""
+        return {
+            "system_fingerprint": response.system_fingerprint,
+        }
+
+    def _get_metadata_from_chat_choice(self, choice: Choice | ChunkChoice) -> dict[str, Any]:
+        """Get metadata from a chat choice."""
+        return {
+            "logprobs": getattr(choice, "logprobs", None),
+        }
+
+    def _get_tool_calls_from_chat_choice(self, choice: Choice | ChunkChoice) -> list[AIContents]:
+        """Get tool calls from a chat choice."""
+        resp: list[AIContents] = []
+        content = choice.message if isinstance(choice, Choice) else choice.delta
+        if content and (tool_calls := getattr(content, "tool_calls", None)) is not None:
+            for tool in cast(list[ChatCompletionMessageToolCall] | list[ChoiceDeltaToolCall], tool_calls):
+                if tool.function:
+                    fcc = FunctionCallContent(
+                        call_id=tool.id if tool.id else "",
+                        name=tool.function.name if tool.function.name else "",
+                        arguments=tool.function.arguments if tool.function.arguments else "",
+                    )
+                    resp.append(fcc)
+
+        # When you enable asynchronous content filtering in Azure OpenAI, you may receive empty deltas
+        return resp
+
+    def _prepare_chat_history_for_request(
+        self,
+        chat_messages: Sequence[ChatMessage],
+        role_key: str = "role",
+        content_key: str = "content",
+    ) -> list[dict[str, Any]]:
+        """Prepare the chat history for a request.
+
+        Allowing customization of the key names for role/author, and optionally overriding the role.
+
+        ChatRole.TOOL messages need to be formatted different than system/user/assistant messages:
+            They require a "tool_call_id" and (function) "name" key, and the "metadata" key should
+            be removed. The "encoding" key should also be removed.
+
+        Override this method to customize the formatting of the chat history for a request.
+
+        Args:
+            chat_messages: The chat history to prepare.
+            role_key: The key name for the role/author.
+            content_key: The key name for the content/message.
+
+        Returns:
+            prepared_chat_history (Any): The prepared chat history for a request.
+        """
+        list_of_list = [self._openai_chat_message_parser(message) for message in chat_messages]
+        # Flatten the list of lists into a single list
+        return list(chain.from_iterable(list_of_list))
+
+    # endregion
+
+    def _openai_chat_message_parser(self, message: ChatMessage) -> list[dict[str, Any]]:
+        """Parse a chat message into the openai format."""
+        all_messages: list[dict[str, Any]] = []
+        args: dict[str, Any] = {
+            "role": message.role.value if isinstance(message.role, ChatRole) else message.role,
+        }
+        if message.additional_properties:
+            args["metadata"] = message.additional_properties
+        for content in message.contents:
+            match content:
+                case FunctionResultContent():
+                    new_args = args.copy()
+                    new_args["tool_call_id"] = content.call_id
+                    new_args["content"] = content.result
+                    all_messages.append(new_args)
+                case FunctionCallContent():
+                    function_call = self._openai_content_parser(content)
+                    if "tool_calls" not in args:
+                        args["tool_calls"] = []
+                    args["tool_calls"].append(function_call)
+                case _:
+                    if "content" not in args:
+                        args["content"] = []
+                    args["content"].append(self._openai_content_parser(content))
+        if "content" in args or "tool_calls" in args:
+            all_messages.append(args)
+        return all_messages
+
+    def _openai_content_parser(self, content: AIContents) -> dict[str, Any]:
+        """Parse contents into the openai format."""
+        match content:
+            case FunctionCallContent():
+                args = json.dumps(content.arguments) if isinstance(content.arguments, Mapping) else content.arguments
+                return {
+                    "id": content.call_id,
+                    "type": "function",
+                    "function": {"name": content.name, "arguments": args},
+                }
+            case FunctionResultContent():
+                return {
+                    "tool_call_id": content.call_id,
+                    "content": content.result,
+                }
+            case _:
+                return content.model_dump(exclude_none=True)
+
+
+class OpenAIChatClient(OpenAIConfigBase, OpenAIChatClientBase):
+    """OpenAI Chat completion class."""
+
+    def __init__(
+        self,
+        ai_model_id: str | None = None,
+        api_key: str | None = None,
+        org_id: str | None = None,
+        default_headers: Mapping[str, str] | None = None,
+        async_client: AsyncOpenAI | None = None,
+        env_file_path: str | None = None,
+        env_file_encoding: str | None = None,
+        instruction_role: str | None = None,
+    ) -> None:
+        """Initialize an OpenAIChatCompletion service.
+
+        Args:
+            ai_model_id (str): OpenAI model name, see
+                https://platform.openai.com/docs/models
+            api_key (str | None): The optional API key to use. If provided will override,
+                the env vars or .env file value.
+            org_id (str | None): The optional org ID to use. If provided will override,
+                the env vars or .env file value.
+            default_headers: The default headers mapping of string keys to
+                string values for HTTP requests. (Optional)
+            async_client (Optional[AsyncOpenAI]): An existing client to use. (Optional)
+            env_file_path (str | None): Use the environment settings file as a fallback
+                to environment variables. (Optional)
+            env_file_encoding (str | None): The encoding of the environment settings file. (Optional)
+            instruction_role (str | None): The role to use for 'instruction' messages, for example,
+        """
+        try:
+            if api_key:
+                openai_settings = OpenAISettings(
+                    api_key=SecretStr(api_key),
+                    org_id=org_id,
+                    chat_model_id=ai_model_id,
+                    env_file_path=env_file_path,
+                    env_file_encoding=env_file_encoding,
+                )
+            else:
+                openai_settings = OpenAISettings(
+                    org_id=org_id,
+                    chat_model_id=ai_model_id,
+                    env_file_path=env_file_path,
+                    env_file_encoding=env_file_encoding,
+                )
+        except ValidationError as ex:
+            raise ServiceInitializationError("Failed to create OpenAI settings.", ex) from ex
+
+        if not async_client and not openai_settings.api_key:
+            raise ServiceInitializationError("The OpenAI API key is required.")
+        if not openai_settings.chat_model_id:
+            raise ServiceInitializationError("The OpenAI model ID is required.")
+
+        super().__init__(
+            ai_model_id=openai_settings.chat_model_id,
+            api_key=openai_settings.api_key.get_secret_value() if openai_settings.api_key else None,
+            org_id=openai_settings.org_id,
+            ai_model_type=OpenAIModelTypes.CHAT,
+            default_headers=default_headers,
+            client=async_client,
+            instruction_role=instruction_role,
+        )
+
+    @classmethod
+    def from_dict(cls, settings: dict[str, Any]) -> "OpenAIChatClient":
+        """Initialize an Open AI service from a dictionary of settings.
+
+        Args:
+            settings: A dictionary of settings for the service.
+        """
+        return OpenAIChatClient(
+            ai_model_id=settings["ai_model_id"],
+            default_headers=settings.get("default_headers"),
+        )

python/packages/main/agent_framework/openai/_shared.py

@@ -0,0 +1,294 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import logging
+from abc import ABC
+from collections.abc import Mapping
+from copy import copy
+from enum import Enum
+from typing import Annotated, Any, ClassVar, Union
+
+from openai import (
+    AsyncOpenAI,
+    AsyncStream,
+    BadRequestError,
+    _legacy_response,  # type: ignore
+)
+from openai.lib._parsing._completions import type_to_response_format_param
+from openai.types import Completion
+from openai.types.audio import Transcription
+from openai.types.chat import ChatCompletion, ChatCompletionChunk
+from openai.types.images_response import ImagesResponse
+from pydantic import BaseModel, ConfigDict, Field, SecretStr, validate_call
+from pydantic.types import StringConstraints
+
+from .._logging import get_logger
+from .._pydantic import AFBaseModel, AFBaseSettings
+from .._types import ChatOptions, SpeechToTextOptions, TextToSpeechOptions
+from ..exceptions import ServiceInitializationError, ServiceInvalidRequestError, ServiceResponseException
+from ..telemetry import APP_INFO, USER_AGENT_KEY, prepend_agent_framework_to_user_agent
+from .exceptions import OpenAIContentFilterException
+
+logger: logging.Logger = get_logger("agent_framework.openai")
+
+
+RESPONSE_TYPE = Union[
+    ChatCompletion,
+    Completion,
+    AsyncStream[ChatCompletionChunk],
+    AsyncStream[Completion],
+    list[Any],
+    ImagesResponse,
+    Transcription,
+    _legacy_response.HttpxBinaryResponseContent,
+]
+
+OPTION_TYPE = Union[
+    ChatOptions,
+    SpeechToTextOptions,
+    TextToSpeechOptions,
+]
+
+
+class OpenAISettings(AFBaseSettings):
+    """OpenAI model settings.
+
+    The settings are first loaded from environment variables with the prefix 'OPENAI_'.
+    If the environment variables are not found, the settings can be loaded from a .env file with the
+    encoding 'utf-8'. If the settings are not found in the .env file, the settings are ignored;
+    however, validation will fail alerting that the settings are missing.
+
+    Optional settings for prefix 'OPENAI_' are:
+    - api_key: SecretStr - OpenAI API key, see https://platform.openai.com/account/api-keys
+        (Env var OPENAI_API_KEY)
+    - org_id: str | None - This is usually optional unless your account belongs to multiple organizations.
+        (Env var OPENAI_ORG_ID)
+    - chat_model_id: str | None - The OpenAI chat model ID to use, for example, gpt-3.5-turbo or gpt-4.
+        (Env var OPENAI_CHAT_MODEL_ID)
+    - responses_model_id: str | None - The OpenAI responses model ID to use, for example, gpt-4o or o1.
+        (Env var OPENAI_RESPONSES_MODEL_ID)
+    - text_model_id: str | None - The OpenAI text model ID to use, for example, gpt-3.5-turbo-instruct.
+        (Env var OPENAI_TEXT_MODEL_ID)
+    - embedding_model_id: str | None - The OpenAI embedding model ID to use, for example, text-embedding-ada-002.
+        (Env var OPENAI_EMBEDDING_MODEL_ID)
+    - text_to_image_model_id: str | None - The OpenAI text to image model ID to use, for example, dall-e-3.
+        (Env var OPENAI_TEXT_TO_IMAGE_MODEL_ID)
+    - audio_to_text_model_id: str | None - The OpenAI audio to text model ID to use, for example, whisper-1.
+        (Env var OPENAI_AUDIO_TO_TEXT_MODEL_ID)
+    - text_to_audio_model_id: str | None - The OpenAI text to audio model ID to use, for example, jukebox-1.
+        (Env var OPENAI_TEXT_TO_AUDIO_MODEL_ID)
+    - realtime_model_id: str | None - The OpenAI realtime model ID to use,
+    for example, gpt-4o-realtime-preview-2024-12-17.
+        (Env var OPENAI_REALTIME_MODEL_ID)
+    - env_file_path: str | None - if provided, the .env settings are read from this file path location
+    """
+
+    env_prefix: ClassVar[str] = "OPENAI_"
+
+    api_key: SecretStr | None = None
+    org_id: str | None = None
+    chat_model_id: str | None = None
+    responses_model_id: str | None = None
+    text_model_id: str | None = None
+    embedding_model_id: str | None = None
+    text_to_image_model_id: str | None = None
+    audio_to_text_model_id: str | None = None
+    text_to_audio_model_id: str | None = None
+    realtime_model_id: str | None = None
+
+
+class OpenAIModelTypes(Enum):
+    """OpenAI model types, can be text, chat or embedding."""
+
+    CHAT = "chat"
+    EMBEDDING = "embedding"
+    TEXT_TO_IMAGE = "text-to-image"
+    SPEECH_TO_TEXT = "speech-to-text"
+    TEXT_TO_SPEECH = "text-to-speech"
+    REALTIME = "realtime"
+    RESPONSE = "response"
+
+
+class OpenAIHandler(AFBaseModel, ABC):
+    """Internal class for calls to OpenAI API's."""
+
+    client: AsyncOpenAI
+    ai_model_id: Annotated[str, StringConstraints(strip_whitespace=True, min_length=1)]
+    ai_model_type: OpenAIModelTypes = OpenAIModelTypes.CHAT
+
+    async def _send_request(self, options: OPTION_TYPE, messages: list[dict[str, Any]] | None = None) -> RESPONSE_TYPE:
+        """Send a request to the OpenAI API."""
+        if self.ai_model_type == OpenAIModelTypes.CHAT:
+            assert isinstance(options, ChatOptions)  # nosec # noqa: S101
+            return await self._send_completion_request(options, messages)
+        # TODO(evmattso): move other PromptExecutionSettings to a common options class
+        if self.ai_model_type == OpenAIModelTypes.EMBEDDING:
+            raise NotImplementedError("Embedding generation is not yet implemented in OpenAIHandler")
+        if self.ai_model_type == OpenAIModelTypes.TEXT_TO_IMAGE:
+            raise NotImplementedError("Text to image generation is not yet implemented in OpenAIHandler")
+        if self.ai_model_type == OpenAIModelTypes.SPEECH_TO_TEXT:
+            assert isinstance(options, SpeechToTextOptions)  # nosec # noqa: S101
+            return await self._send_audio_to_text_request(options)
+        if self.ai_model_type == OpenAIModelTypes.TEXT_TO_SPEECH:
+            assert isinstance(options, TextToSpeechOptions)  # nosec # noqa: S101
+            return await self._send_text_to_audio_request(options)
+
+        raise NotImplementedError(f"Model type {self.ai_model_type} is not supported")
+
+    async def _send_completion_request(
+        self,
+        chat_options: "ChatOptions",
+        messages: list[dict[str, Any]] | None = None,
+    ) -> ChatCompletion | AsyncStream[ChatCompletionChunk]:
+        """Execute the appropriate call to OpenAI models."""
+        try:
+            options_dict = chat_options.to_provider_settings()
+            if messages and "messages" not in options_dict:
+                options_dict["messages"] = messages
+            if "messages" not in options_dict:
+                raise ServiceInvalidRequestError("Messages are required for chat completions")
+            self._handle_structured_outputs(chat_options, options_dict)
+            if chat_options.tools is None:
+                options_dict.pop("parallel_tool_calls", None)
+            return await self.client.chat.completions.create(**options_dict)  # type: ignore
+        except BadRequestError as ex:
+            if ex.code == "content_filter":
+                raise OpenAIContentFilterException(
+                    f"{type(self)} service encountered a content error",
+                    ex,
+                ) from ex
+            raise ServiceResponseException(
+                f"{type(self)} service failed to complete the prompt",
+                ex,
+            ) from ex
+        except Exception as ex:
+            raise ServiceResponseException(
+                f"{type(self)} service failed to complete the prompt",
+                ex,
+            ) from ex
+
+    async def _send_audio_to_text_request(self, options: SpeechToTextOptions) -> Transcription:
+        """Send a request to the OpenAI audio to text endpoint."""
+        if not options.additional_properties["filename"]:
+            raise ServiceInvalidRequestError("Audio file is required for audio to text service")
+
+        try:
+            # TODO(peterychang): open isn't async safe
+            with open(options.additional_properties["filename"], "rb") as audio_file:  # noqa: ASYNC230
+                return await self.client.audio.transcriptions.create(  # type: ignore
+                    file=audio_file,
+                    **options.to_provider_settings(exclude={"filename"}),
+                )
+        except Exception as ex:
+            raise ServiceResponseException(
+                f"{type(self)} service failed to transcribe audio",
+                ex,
+            ) from ex
+
+    async def _send_text_to_audio_request(
+        self, options: TextToSpeechOptions
+    ) -> _legacy_response.HttpxBinaryResponseContent:
+        """Send a request to the OpenAI text to audio endpoint.
+
+        The OpenAI API returns the content of the generated audio file.
+        """
+        try:
+            return await self.client.audio.speech.create(
+                **options.to_provider_settings(),
+            )
+        except Exception as ex:
+            raise ServiceResponseException(
+                f"{type(self)} service failed to generate audio",
+                ex,
+            ) from ex
+
+    def _handle_structured_outputs(self, chat_options: "ChatOptions", options_dict: dict[str, Any]) -> None:
+        if (
+            chat_options.response_format
+            and isinstance(chat_options.response_format, type)
+            and issubclass(chat_options.response_format, BaseModel)
+        ):
+            options_dict["response_format"] = type_to_response_format_param(chat_options.response_format)
+
+
+class OpenAIConfigBase(OpenAIHandler):
+    """Internal class for configuring a connection to an OpenAI service."""
+
+    @validate_call(config=ConfigDict(arbitrary_types_allowed=True))
+    def __init__(
+        self,
+        ai_model_id: str = Field(min_length=1),
+        api_key: str | None = Field(min_length=1),
+        ai_model_type: OpenAIModelTypes | None = OpenAIModelTypes.CHAT,
+        org_id: str | None = None,
+        default_headers: Mapping[str, str] | None = None,
+        client: AsyncOpenAI | None = None,
+        instruction_role: str | None = None,
+        **kwargs: Any,
+    ) -> None:
+        """Initialize a client for OpenAI services.
+
+        This constructor sets up a client to interact with OpenAI's API, allowing for
+        different types of AI model interactions, like chat or text completion.
+
+        Args:
+            ai_model_id (str): OpenAI model identifier. Must be non-empty.
+                Default to a preset value.
+            api_key (str): OpenAI API key for authentication.
+                Must be non-empty. (Optional)
+            ai_model_type (OpenAIModelTypes): The type of OpenAI
+                model to interact with. Defaults to CHAT.
+            org_id (str): OpenAI organization ID. This is optional
+                unless the account belongs to multiple organizations.
+            default_headers (Mapping[str, str]): Default headers
+                for HTTP requests. (Optional)
+            client (AsyncOpenAI): An existing OpenAI client, optional.
+            instruction_role (str): The role to use for 'instruction'
+                messages, for example, summarization prompts could use `developer` or `system`. (Optional)
+            kwargs: Additional keyword arguments.
+
+        """
+        # Merge APP_INFO into the headers if it exists
+        merged_headers = dict(copy(default_headers)) if default_headers else {}
+        if APP_INFO:
+            merged_headers.update(APP_INFO)
+            merged_headers = prepend_agent_framework_to_user_agent(merged_headers)
+
+        if not client:
+            if not api_key:
+                raise ServiceInitializationError("Please provide an api_key")
+            client = AsyncOpenAI(
+                api_key=api_key,
+                organization=org_id,
+                default_headers=merged_headers,
+            )
+        args = {
+            "ai_model_id": ai_model_id,
+            "client": client,
+            "ai_model_type": ai_model_type,
+        }
+        if instruction_role:
+            args["instruction_role"] = instruction_role
+        super().__init__(**args, **kwargs)
+
+    def to_dict(self) -> dict[str, Any]:
+        """Create a dict of the service settings."""
+        client_settings = {
+            "api_key": self.client.api_key,
+            "default_headers": {k: v for k, v in self.client.default_headers.items() if k != USER_AGENT_KEY},
+        }
+        if self.client.organization:
+            client_settings["org_id"] = self.client.organization
+        base = self.model_dump(
+            exclude={
+                "prompt_tokens",
+                "completion_tokens",
+                "total_tokens",
+                "api_type",
+                "ai_model_type",
+                "client",
+            },
+            by_alias=True,
+            exclude_none=True,
+        )
+        base.update(client_settings)
+        return base

python/packages/main/agent_framework/openai/exceptions.py

@@ -0,0 +1,90 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+from openai import BadRequestError
+
+from ..exceptions import ServiceContentFilterException
+
+
+class ContentFilterResultSeverity(Enum):
+    """The severity of the content filter result."""
+
+    HIGH = "high"
+    MEDIUM = "medium"
+    SAFE = "safe"
+    LOW = "low"
+
+
+@dataclass
+class ContentFilterResult:
+    """The result of a content filter check."""
+
+    filtered: bool = False
+    detected: bool = False
+    severity: ContentFilterResultSeverity = ContentFilterResultSeverity.SAFE
+
+    @classmethod
+    def from_inner_error_result(cls, inner_error_results: dict[str, Any]) -> "ContentFilterResult":
+        """Creates a ContentFilterResult from the inner error results.
+
+        Args:
+            key (str): The key to get the inner error result from.
+            inner_error_results (Dict[str, Any]): The inner error results.
+
+        Returns:
+            ContentFilterResult: The ContentFilterResult.
+        """
+        return cls(
+            filtered=inner_error_results.get("filtered", False),
+            detected=inner_error_results.get("detected", False),
+            severity=ContentFilterResultSeverity(
+                inner_error_results.get("severity", ContentFilterResultSeverity.SAFE.value)
+            ),
+        )
+
+
+class ContentFilterCodes(Enum):
+    """Content filter codes."""
+
+    RESPONSIBLE_AI_POLICY_VIOLATION = "ResponsibleAIPolicyViolation"
+
+
+@dataclass
+class OpenAIContentFilterException(ServiceContentFilterException):
+    """AI exception for an error from Azure OpenAI's content filter."""
+
+    # The parameter that caused the error.
+    param: str | None
+
+    # The error code specific to the content filter.
+    content_filter_code: ContentFilterCodes
+
+    # The results of the different content filter checks.
+    content_filter_result: dict[str, ContentFilterResult]
+
+    def __init__(
+        self,
+        message: str,
+        inner_exception: BadRequestError,
+    ) -> None:
+        """Initializes a new instance of the ContentFilterAIException class.
+
+        Args:
+            message (str): The error message.
+            inner_exception (Exception): The inner exception.
+        """
+        super().__init__(message)
+
+        self.param = inner_exception.param
+        if inner_exception.body is not None and isinstance(inner_exception.body, dict):
+            inner_error = inner_exception.body.get("innererror", {})  # type: ignore
+            self.content_filter_code = ContentFilterCodes(
+                inner_error.get("code", ContentFilterCodes.RESPONSIBLE_AI_POLICY_VIOLATION.value)  # type: ignore
+            )
+            self.content_filter_result = {
+                key: ContentFilterResult.from_inner_error_result(values)  # type: ignore
+                for key, values in inner_error.get("content_filter_result", {}).items()  # type: ignore
+            }

python/packages/main/agent_framework/telemetry.py

@@ -0,0 +1,46 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import os
+from importlib.metadata import PackageNotFoundError, version
+from typing import Any, Final
+
+try:
+    version_info = version("agent-framework")
+except PackageNotFoundError:
+    version_info = "dev"
+
+# Note that if this environment variable does not exist, telemetry is enabled.
+TELEMETRY_DISABLED_ENV_VAR = "AZURE_TELEMETRY_DISABLED"
+IS_TELEMETRY_ENABLED = os.environ.get(TELEMETRY_DISABLED_ENV_VAR, "false").lower() not in ["true", "1"]
+
+APP_INFO = (
+    {
+        "agent-framework-version": f"python/{version_info}",
+    }
+    if IS_TELEMETRY_ENABLED
+    else None
+)
+USER_AGENT_KEY: Final[str] = "User-Agent"
+HTTP_USER_AGENT: Final[str] = "agent-framework-python"
+AGENT_FRAMEWORK_USER_AGENT = f"{HTTP_USER_AGENT}/{version_info}"
+
+
+def prepend_agent_framework_to_user_agent(headers: dict[str, Any]) -> dict[str, Any]:
+    """Prepend "agent-framework" to the User-Agent in the headers.
+
+    Args:
+        headers: The existing headers dictionary.
+
+    Returns:
+        The modified headers dictionary with "agent-framework-python/{version}" prepended to the User-Agent.
+    """
+    headers[USER_AGENT_KEY] = (
+        f"{AGENT_FRAMEWORK_USER_AGENT} {headers[USER_AGENT_KEY]}"
+        if USER_AGENT_KEY in headers
+        else AGENT_FRAMEWORK_USER_AGENT
+    )
+
+    return headers
+
+
+__all__ = ["AGENT_FRAMEWORK_USER_AGENT", "APP_INFO", "USER_AGENT_KEY", "prepend_agent_framework_to_user_agent"]