Merge branch 'main' into feature-azure-functions

This commit is contained in:
Dmytro Struk
2025-11-04 16:44:58 -08:00
Unverified
147 changed files with 9276 additions and 4903 deletions
+25 -1
View File
@@ -7,6 +7,29 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
## [1.0.0b251104] - 2025-11-04
### Added
- Introducing the Anthropic Client ([#1819](https://github.com/microsoft/agent-framework/pull/1819))
### Changed
- [BREAKING] Consolidate workflow run APIs ([#1723](https://github.com/microsoft/agent-framework/pull/1723))
- [BREAKING] Remove request_type param from ctx.request_info() ([#1824](https://github.com/microsoft/agent-framework/pull/1824))
- [BREAKING] Cleanup of dependencies ([#1803](https://github.com/microsoft/agent-framework/pull/1803))
- [BREAKING] Replace `RequestInfoExecutor` with `request_info` API and `@response_handler` ([#1466](https://github.com/microsoft/agent-framework/pull/1466))
- Azure AI Search Support Update + Refactored Samples & Unit Tests ([#1683](https://github.com/microsoft/agent-framework/pull/1683))
- Lab: Updates to GAIA module ([#1763](https://github.com/microsoft/agent-framework/pull/1763))
### Fixed
- Azure AI `top_p` and `temperature` parameters fix ([#1839](https://github.com/microsoft/agent-framework/pull/1839))
- Ensure agent thread is part of checkpoint ([#1756](https://github.com/microsoft/agent-framework/pull/1756))
- Fix middleware and cleanup confusing function ([#1865](https://github.com/microsoft/agent-framework/pull/1865))
- Fix type compatibility check ([#1753](https://github.com/microsoft/agent-framework/pull/1753))
- Fix mcp tool cloning for handoff pattern ([#1883](https://github.com/microsoft/agent-framework/pull/1883))
## [1.0.0b251028] - 2025-10-28
### Added
@@ -124,7 +147,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
For more information, see the [announcement blog post](https://devblogs.microsoft.com/foundry/introducing-microsoft-agent-framework-the-open-source-engine-for-agentic-ai-apps/).
[Unreleased]: https://github.com/microsoft/agent-framework/compare/python-1.0.0b251028...HEAD
[Unreleased]: https://github.com/microsoft/agent-framework/compare/python-1.0.0b251104...HEAD
[1.0.0b251104]: https://github.com/microsoft/agent-framework/compare/python-1.0.0b251028...python-1.0.0b251104
[1.0.0b251028]: https://github.com/microsoft/agent-framework/compare/python-1.0.0b251016...python-1.0.0b251028
[1.0.0b251016]: https://github.com/microsoft/agent-framework/compare/python-1.0.0b251007...python-1.0.0b251016
[1.0.0b251007]: https://github.com/microsoft/agent-framework/compare/python-1.0.0b251001...python-1.0.0b251007
@@ -127,7 +127,21 @@ class A2AAgent(BaseAgent):
)
factory = ClientFactory(config)
interceptors = [auth_interceptor] if auth_interceptor is not None else None
self.client = factory.create(agent_card, interceptors=interceptors) # type: ignore
# Attempt transport negotiation with the provided agent card
try:
self.client = factory.create(agent_card, interceptors=interceptors) # type: ignore
except Exception as transport_error:
# Transport negotiation failed - fall back to minimal agent card with JSONRPC
fallback_card = minimal_agent_card(agent_card.url, [TransportProtocol.jsonrpc])
try:
self.client = factory.create(fallback_card, interceptors=interceptors) # type: ignore
except Exception as fallback_error:
raise RuntimeError(
f"A2A transport negotiation failed. "
f"Primary error: {transport_error}. "
f"Fallback error: {fallback_error}"
) from transport_error
async def __aenter__(self) -> "A2AAgent":
"""Async context manager entry."""
+1 -1
View File
@@ -4,7 +4,7 @@ description = "A2A integration for Microsoft Agent Framework."
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
readme = "README.md"
requires-python = ">=3.10"
version = "1.0.0b251028"
version = "1.0.0b251104"
license-files = ["LICENSE"]
urls.homepage = "https://aka.ms/agent-framework"
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
+41 -2
View File
@@ -2,10 +2,22 @@
from collections.abc import AsyncIterator
from typing import Any
from unittest.mock import AsyncMock, MagicMock
from unittest.mock import AsyncMock, MagicMock, patch
from uuid import uuid4
from a2a.types import Artifact, DataPart, FilePart, FileWithUri, Message, Part, Task, TaskState, TaskStatus, TextPart
from a2a.types import (
AgentCard,
Artifact,
DataPart,
FilePart,
FileWithUri,
Message,
Part,
Task,
TaskState,
TaskStatus,
TextPart,
)
from a2a.types import Role as A2ARole
from agent_framework import (
AgentRunResponse,
@@ -515,3 +527,30 @@ def test_auth_interceptor_parameter() -> None:
# Verify the agent was created successfully
assert agent.name == "test-agent"
assert agent.client is not None
def test_transport_negotiation_both_fail() -> None:
"""Test that RuntimeError is raised when both primary and fallback transport negotiation fail."""
# Create a mock agent card
mock_agent_card = MagicMock(spec=AgentCard)
mock_agent_card.url = "http://test-agent.example.com"
# Mock the factory to simulate both primary and fallback failures
mock_factory = MagicMock()
# Both calls to factory.create() fail
primary_error = Exception("no compatible transports found")
fallback_error = Exception("fallback also failed")
mock_factory.create.side_effect = [primary_error, fallback_error]
with (
patch("agent_framework_a2a._agent.ClientFactory", return_value=mock_factory),
patch("agent_framework_a2a._agent.minimal_agent_card"),
patch("agent_framework_a2a._agent.httpx.AsyncClient"),
raises(RuntimeError, match="A2A transport negotiation failed"),
):
# Attempt to create A2AAgent - should raise RuntimeError
A2AAgent(
name="test-agent",
agent_card=mock_agent_card,
)
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) Microsoft Corporation.
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE
+18
View File
@@ -0,0 +1,18 @@
# Get Started with Microsoft Agent Framework Anthropic
Please install this package via pip:
```bash
pip install agent-framework-anthropic --pre
```
## Anthropic Integration
The Anthropic integration enables communication with the Anthropic API, allowing your Agent Framework applications to leverage Anthropic's capabilities.
### Basic Usage Example
See the [Anthropic agent examples](https://github.com/microsoft/agent-framework/tree/main/python/samples/getting_started/agents/anthropic/) which demonstrate:
- Connecting to a Anthropic endpoint with an agent
- Streaming and non-streaming responses
@@ -0,0 +1,15 @@
# Copyright (c) Microsoft. All rights reserved.
import importlib.metadata
from ._chat_client import AnthropicClient
try:
__version__ = importlib.metadata.version(__name__)
except importlib.metadata.PackageNotFoundError:
__version__ = "0.0.0" # Fallback for development mode
__all__ = [
"AnthropicClient",
"__version__",
]
@@ -0,0 +1,658 @@
# Copyright (c) Microsoft. All rights reserved.
from collections.abc import AsyncIterable, MutableMapping, MutableSequence, Sequence
from typing import Any, ClassVar, Final, TypeVar
from agent_framework import (
AGENT_FRAMEWORK_USER_AGENT,
AIFunction,
Annotations,
BaseChatClient,
ChatMessage,
ChatOptions,
ChatResponse,
ChatResponseUpdate,
CitationAnnotation,
Contents,
FinishReason,
FunctionCallContent,
FunctionResultContent,
HostedCodeInterpreterTool,
HostedMCPTool,
HostedWebSearchTool,
Role,
TextContent,
TextReasoningContent,
TextSpanRegion,
ToolProtocol,
UsageContent,
UsageDetails,
get_logger,
prepare_function_call_results,
use_chat_middleware,
use_function_invocation,
)
from agent_framework._pydantic import AFBaseSettings
from agent_framework.exceptions import ServiceInitializationError
from agent_framework.observability import use_observability
from anthropic import AsyncAnthropic
from anthropic.types.beta import (
BetaContentBlock,
BetaMessage,
BetaMessageDeltaUsage,
BetaRawContentBlockDelta,
BetaRawMessageStreamEvent,
BetaTextBlock,
BetaUsage,
)
from pydantic import SecretStr, ValidationError
logger = get_logger("agent_framework.anthropic")
ANTHROPIC_DEFAULT_MAX_TOKENS: Final[int] = 1024
BETA_FLAGS: Final[list[str]] = ["mcp-client-2025-04-04", "code-execution-2025-08-25"]
ROLE_MAP: dict[Role, str] = {
Role.USER: "user",
Role.ASSISTANT: "assistant",
Role.SYSTEM: "user",
Role.TOOL: "user",
}
FINISH_REASON_MAP: dict[str, FinishReason] = {
"stop_sequence": FinishReason.STOP,
"max_tokens": FinishReason.LENGTH,
"tool_use": FinishReason.TOOL_CALLS,
"end_turn": FinishReason.STOP,
"refusal": FinishReason.CONTENT_FILTER,
"pause_turn": FinishReason.STOP,
}
class AnthropicSettings(AFBaseSettings):
"""Anthropic Project settings.
The settings are first loaded from environment variables with the prefix 'ANTHROPIC_'.
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.
Keyword Args:
api_key: The Anthropic API key.
chat_model_id: The Anthropic chat model ID.
env_file_path: If provided, the .env settings are read from this file path location.
env_file_encoding: The encoding of the .env file, defaults to 'utf-8'.
Examples:
.. code-block:: python
from agent_framework.anthropic import AnthropicSettings
# Using environment variables
# Set ANTHROPIC_API_KEY=your_anthropic_api_key
# ANTHROPIC_CHAT_MODEL_ID=claude-sonnet-4-5-20250929
# Or passing parameters directly
settings = AnthropicSettings(chat_model_id="claude-sonnet-4-5-20250929")
# Or loading from a .env file
settings = AnthropicSettings(env_file_path="path/to/.env")
"""
env_prefix: ClassVar[str] = "ANTHROPIC_"
api_key: SecretStr | None = None
chat_model_id: str | None = None
TAnthropicClient = TypeVar("TAnthropicClient", bound="AnthropicClient")
@use_function_invocation
@use_observability
@use_chat_middleware
class AnthropicClient(BaseChatClient):
"""Anthropic Chat client."""
OTEL_PROVIDER_NAME: ClassVar[str] = "anthropic" # type: ignore[reportIncompatibleVariableOverride, misc]
def __init__(
self,
*,
api_key: str | None = None,
model_id: str | None = None,
anthropic_client: AsyncAnthropic | None = None,
env_file_path: str | None = None,
env_file_encoding: str | None = None,
**kwargs: Any,
) -> None:
"""Initialize an Anthropic Agent client.
Keyword Args:
api_key: The Anthropic API key to use for authentication.
model_id: The ID of the model to use.
anthropic_client: An existing Anthropic client to use. If not provided, one will be created.
This can be used to further configure the client before passing it in.
For instance if you need to set a different base_url for testing or private deployments.
env_file_path: Path to environment file for loading settings.
env_file_encoding: Encoding of the environment file.
kwargs: Additional keyword arguments passed to the parent class.
Examples:
.. code-block:: python
from agent_framework.anthropic import AnthropicClient
from azure.identity.aio import DefaultAzureCredential
# Using environment variables
# Set ANTHROPIC_API_KEY=your_anthropic_api_key
# ANTHROPIC_CHAT_MODEL_ID=claude-sonnet-4-5-20250929
# Or passing parameters directly
client = AnthropicClient(
model_id="claude-sonnet-4-5-20250929",
api_key="your_anthropic_api_key",
)
# Or loading from a .env file
client = AnthropicClient(env_file_path="path/to/.env")
# Or passing in an existing client
from anthropic import AsyncAnthropic
anthropic_client = AsyncAnthropic(
api_key="your_anthropic_api_key", base_url="https://custom-anthropic-endpoint.com"
)
client = AnthropicClient(
model_id="claude-sonnet-4-5-20250929",
anthropic_client=anthropic_client,
)
"""
try:
anthropic_settings = AnthropicSettings(
api_key=api_key, # type: ignore[arg-type]
chat_model_id=model_id,
env_file_path=env_file_path,
env_file_encoding=env_file_encoding,
)
except ValidationError as ex:
raise ServiceInitializationError("Failed to create Anthropic settings.", ex) from ex
if anthropic_client is None:
if not anthropic_settings.api_key:
raise ServiceInitializationError(
"Anthropic API key is required. Set via 'api_key' parameter "
"or 'ANTHROPIC_API_KEY' environment variable."
)
anthropic_client = AsyncAnthropic(
api_key=anthropic_settings.api_key.get_secret_value(),
default_headers={"User-Agent": AGENT_FRAMEWORK_USER_AGENT},
)
# Initialize parent
super().__init__(**kwargs)
# Initialize instance variables
self.anthropic_client = anthropic_client
self.model_id = anthropic_settings.chat_model_id
# streaming requires tracking the last function call ID and name
self._last_call_id_name: tuple[str, str] | None = None
# region Get response methods
async def _inner_get_response(
self,
*,
messages: MutableSequence[ChatMessage],
chat_options: ChatOptions,
**kwargs: Any,
) -> ChatResponse:
# Extract necessary state from messages and options
run_options = self._create_run_options(messages, chat_options, **kwargs)
message = await self.anthropic_client.beta.messages.create(**run_options, stream=False)
return self._process_message(message)
async def _inner_get_streaming_response(
self,
*,
messages: MutableSequence[ChatMessage],
chat_options: ChatOptions,
**kwargs: Any,
) -> AsyncIterable[ChatResponseUpdate]:
# Extract necessary state from messages and options
run_options = self._create_run_options(messages, chat_options, **kwargs)
async for chunk in await self.anthropic_client.beta.messages.create(**run_options, stream=True):
parsed_chunk = self._process_stream_event(chunk)
if parsed_chunk:
yield parsed_chunk
# region Create Run Options and Helpers
def _create_run_options(
self,
messages: MutableSequence[ChatMessage],
chat_options: ChatOptions,
**kwargs: Any,
) -> dict[str, Any]:
"""Create run options for the Anthropic client based on messages and chat options.
Args:
messages: The list of chat messages.
chat_options: The chat options.
kwargs: Additional keyword arguments.
Returns:
A dictionary of run options for the Anthropic client.
"""
run_options: dict[str, Any] = {
"model": chat_options.model_id or self.model_id,
"messages": self._convert_messages_to_anthropic_format(messages),
"max_tokens": chat_options.max_tokens or ANTHROPIC_DEFAULT_MAX_TOKENS,
"extra_headers": {"User-Agent": AGENT_FRAMEWORK_USER_AGENT},
"betas": BETA_FLAGS,
}
# Add any additional options from chat_options or kwargs
if chat_options.temperature is not None:
run_options["temperature"] = chat_options.temperature
if chat_options.top_p is not None:
run_options["top_p"] = chat_options.top_p
if chat_options.stop is not None:
run_options["stop_sequences"] = chat_options.stop
if messages and isinstance(messages[0], ChatMessage) and messages[0].role == Role.SYSTEM:
# first system message is passed as instructions
run_options["system"] = messages[0].text
if chat_options.tool_choice is not None:
match (
chat_options.tool_choice if isinstance(chat_options.tool_choice, str) else chat_options.tool_choice.mode
):
case "auto":
run_options["tool_choice"] = {"type": "auto"}
if chat_options.allow_multiple_tool_calls is not None:
run_options["tool_choice"][ # type:ignore[reportArgumentType]
"disable_parallel_tool_use"
] = not chat_options.allow_multiple_tool_calls
case "required":
if chat_options.tool_choice.required_function_name:
run_options["tool_choice"] = {
"type": "tool",
"name": chat_options.tool_choice.required_function_name,
}
if chat_options.allow_multiple_tool_calls is not None:
run_options["tool_choice"][ # type:ignore[reportArgumentType]
"disable_parallel_tool_use"
] = not chat_options.allow_multiple_tool_calls
else:
run_options["tool_choice"] = {"type": "any"}
if chat_options.allow_multiple_tool_calls is not None:
run_options["tool_choice"][ # type:ignore[reportArgumentType]
"disable_parallel_tool_use"
] = not chat_options.allow_multiple_tool_calls
case "none":
run_options["tool_choice"] = {"type": "none"}
case _:
logger.debug(f"Ignoring unsupported tool choice mode: {chat_options.tool_choice.mode} for now")
if tools_and_mcp := self._convert_tools_to_anthropic_format(chat_options.tools):
run_options.update(tools_and_mcp)
if chat_options.additional_properties:
run_options.update(chat_options.additional_properties)
run_options.update(kwargs)
return run_options
def _convert_messages_to_anthropic_format(self, messages: MutableSequence[ChatMessage]) -> list[dict[str, Any]]:
"""Convert a list of ChatMessages to the format expected by the Anthropic client.
This skips the first message if it is a system message,
as Anthropic expects system instructions as a separate parameter.
"""
# first system message is passed as instructions
if messages and isinstance(messages[0], ChatMessage) and messages[0].role == Role.SYSTEM:
return [self._convert_message_to_anthropic_format(msg) for msg in messages[1:]]
return [self._convert_message_to_anthropic_format(msg) for msg in messages]
def _convert_message_to_anthropic_format(self, message: ChatMessage) -> dict[str, Any]:
"""Convert a ChatMessage to the format expected by the Anthropic client.
Args:
message: The ChatMessage to convert.
Returns:
A dictionary representing the message in Anthropic format.
"""
a_content: list[dict[str, Any]] = []
for content in message.contents:
match content.type:
case "text":
a_content.append({"type": "text", "text": content.text})
case "data":
if content.has_top_level_media_type("image"):
a_content.append({
"type": "image",
"source": {"data": content.uri, "media_type": content.media_type},
})
case "uri":
if content.has_top_level_media_type("image"):
a_content.append({"type": "image", "source": {"type": "url", "url": content.uri}})
case "function_call":
a_content.append({
"type": "tool_use",
"id": content.call_id,
"name": content.name,
"input": content.parse_arguments(),
})
case "function_result":
a_content.append({
"type": "tool_result",
"tool_use_id": content.call_id,
"content": prepare_function_call_results(content.result),
"is_error": content.exception is not None,
})
case "text_reasoning":
a_content.append({"type": "thinking", "thinking": content.text})
case _:
logger.debug(f"Ignoring unsupported content type: {content.type} for now")
return {
"role": ROLE_MAP.get(message.role, "user"),
"content": a_content,
}
def _convert_tools_to_anthropic_format(
self, tools: list[ToolProtocol | MutableMapping[str, Any]] | None
) -> dict[str, Any] | None:
if not tools:
return None
tool_list: list[MutableMapping[str, Any]] = []
mcp_server_list: list[MutableMapping[str, Any]] = []
for tool in tools:
match tool:
case MutableMapping():
tool_list.append(tool)
case AIFunction():
tool_list.append({
"type": "custom",
"name": tool.name,
"description": tool.description,
"input_schema": tool.parameters(),
})
case HostedWebSearchTool():
search_tool: dict[str, Any] = {
"type": "web_search_20250305",
"name": "web_search",
}
if tool.additional_properties:
search_tool.update(tool.additional_properties)
tool_list.append(search_tool)
case HostedCodeInterpreterTool():
code_tool: dict[str, Any] = {
"type": "code_execution_20250825",
"name": "code_interpreter",
}
tool_list.append(code_tool)
case HostedMCPTool():
server_def: dict[str, Any] = {
"type": "url",
"name": tool.name,
"url": str(tool.url),
}
if tool.allowed_tools:
server_def["tool_configuration"] = {"allowed_tools": list(tool.allowed_tools)}
if tool.headers and (auth := tool.headers.get("authorization")):
server_def["authorization_token"] = auth
mcp_server_list.append(server_def)
case _:
logger.debug(f"Ignoring unsupported tool type: {type(tool)} for now")
all_tools: dict[str, list[MutableMapping[str, Any]]] = {}
if tool_list:
all_tools["tools"] = tool_list
if mcp_server_list:
all_tools["mcp_servers"] = mcp_server_list
return all_tools
# region Response Processing Methods
def _process_message(self, message: BetaMessage) -> ChatResponse:
"""Process the response from the Anthropic client.
Args:
message: The message returned by the Anthropic client.
Returns:
A ChatResponse object containing the processed response.
"""
return ChatResponse(
response_id=message.id,
messages=[
ChatMessage(
role=Role.ASSISTANT,
contents=self._parse_message_contents(message.content),
raw_representation=message,
)
],
usage_details=self._parse_message_usage(message.usage),
model_id=message.model,
finish_reason=FINISH_REASON_MAP.get(message.stop_reason) if message.stop_reason else None,
raw_response=message,
)
def _process_stream_event(self, event: BetaRawMessageStreamEvent) -> ChatResponseUpdate | None:
"""Process a streaming event from the Anthropic client.
Args:
event: The streaming event returned by the Anthropic client.
Returns:
A ChatResponseUpdate object containing the processed update.
"""
match event.type:
case "message_start":
usage_details: list[UsageContent] = []
if event.message.usage and (details := self._parse_message_usage(event.message.usage)):
usage_details.append(UsageContent(details=details))
return ChatResponseUpdate(
response_id=event.message.id,
contents=[*self._parse_message_contents(event.message.content), *usage_details],
model_id=event.message.model,
finish_reason=FINISH_REASON_MAP.get(event.message.stop_reason)
if event.message.stop_reason
else None,
raw_response=event,
)
case "message_delta":
usage = self._parse_message_usage(event.usage)
return ChatResponseUpdate(
contents=[UsageContent(details=usage, raw_representation=event.usage)] if usage else [],
raw_response=event,
)
case "message_stop":
logger.debug("Received message_stop event; no content to process.")
case "content_block_start":
contents = self._parse_message_contents([event.content_block])
return ChatResponseUpdate(
contents=contents,
raw_response=event,
)
case "content_block_delta":
contents = self._parse_message_contents([event.delta])
return ChatResponseUpdate(
contents=contents,
raw_response=event,
)
case "content_block_stop":
logger.debug("Received content_block_stop event; no content to process.")
case _:
logger.debug(f"Ignoring unsupported event type: {event.type}")
return None
def _parse_message_usage(self, usage: BetaUsage | BetaMessageDeltaUsage | None) -> UsageDetails | None:
"""Parse usage details from the Anthropic message usage."""
if not usage:
return None
usage_details = UsageDetails(output_token_count=usage.output_tokens)
if usage.input_tokens is not None:
usage_details.input_token_count = usage.input_tokens
if usage.cache_creation_input_tokens is not None:
usage_details.additional_counts["anthropic.cache_creation_input_tokens"] = usage.cache_creation_input_tokens
if usage.cache_read_input_tokens is not None:
usage_details.additional_counts["anthropic.cache_read_input_tokens"] = usage.cache_read_input_tokens
return usage_details
def _parse_message_contents(
self, content: Sequence[BetaContentBlock | BetaRawContentBlockDelta | BetaTextBlock]
) -> list[Contents]:
"""Parse contents from the Anthropic message."""
contents: list[Contents] = []
for content_block in content:
match content_block.type:
case "text" | "text_delta":
contents.append(
TextContent(
text=content_block.text,
raw_representation=content_block,
annotations=self._parse_citations(content_block),
)
)
case "tool_use":
self._last_call_id_name = (content_block.id, content_block.name)
contents.append(
FunctionCallContent(
call_id=content_block.id,
name=content_block.name,
arguments=content_block.input,
raw_representation=content_block,
)
)
case "mcp_tool_use" | "server_tool_use":
self._last_call_id_name = (content_block.id, content_block.name)
contents.append(
FunctionCallContent(
call_id=content_block.id,
name=content_block.name,
arguments=content_block.input,
raw_representation=content_block,
)
)
case "mcp_tool_result":
call_id, name = self._last_call_id_name or (None, None)
contents.append(
FunctionResultContent(
call_id=content_block.tool_use_id,
name=name if name and call_id == content_block.tool_use_id else "mcp_tool",
result=self._parse_message_contents(content_block.content)
if isinstance(content_block.content, list)
else content_block.content,
raw_representation=content_block,
)
)
case "web_search_tool_result" | "web_fetch_tool_result":
call_id, name = self._last_call_id_name or (None, None)
contents.append(
FunctionResultContent(
call_id=content_block.tool_use_id,
name=name if name and call_id == content_block.tool_use_id else "web_tool",
result=content_block.content,
raw_representation=content_block,
)
)
case (
"code_execution_tool_result"
| "bash_code_execution_tool_result"
| "text_editor_code_execution_tool_result"
):
call_id, name = self._last_call_id_name or (None, None)
contents.append(
FunctionResultContent(
call_id=content_block.tool_use_id,
name=name if name and call_id == content_block.tool_use_id else "code_execution_tool",
result=content_block.content,
raw_representation=content_block,
)
)
case "input_json_delta":
call_id, name = self._last_call_id_name if self._last_call_id_name else ("", "")
contents.append(
FunctionCallContent(
call_id=call_id,
name=name,
arguments=content_block.partial_json,
raw_representation=content_block,
)
)
case "thinking" | "thinking_delta":
contents.append(TextReasoningContent(text=content_block.thinking, raw_representation=content_block))
case _:
logger.debug(f"Ignoring unsupported content type: {content_block.type} for now")
return contents
def _parse_citations(
self, content_block: BetaContentBlock | BetaRawContentBlockDelta | BetaTextBlock
) -> list[Annotations] | None:
content_citations = getattr(content_block, "citations", None)
if not content_citations:
return None
annotations: list[Annotations] = []
for citation in content_citations:
cit = CitationAnnotation(raw_representation=citation)
match citation.type:
case "char_location":
cit.title = citation.title
cit.snippet = citation.cited_text
if citation.file_id:
cit.file_id = citation.file_id
if not cit.annotated_regions:
cit.annotated_regions = []
cit.annotated_regions.append(
TextSpanRegion(start_index=citation.start_char_index, end_index=citation.end_char_index)
)
case "page_location":
cit.title = citation.document_title
cit.snippet = citation.cited_text
if citation.file_id:
cit.file_id = citation.file_id
if not cit.annotated_regions:
cit.annotated_regions = []
cit.annotated_regions.append(
TextSpanRegion(
start_index=citation.start_page_number,
end_index=citation.end_page_number,
)
)
case "content_block_location":
cit.title = citation.document_title
cit.snippet = citation.cited_text
if citation.file_id:
cit.file_id = citation.file_id
if not cit.annotated_regions:
cit.annotated_regions = []
cit.annotated_regions.append(
TextSpanRegion(start_index=citation.start_block_index, end_index=citation.end_block_index)
)
case "web_search_result_location":
cit.title = citation.title
cit.snippet = citation.cited_text
cit.url = citation.url
case "search_result_location":
cit.title = citation.title
cit.snippet = citation.cited_text
cit.url = citation.source
if not cit.annotated_regions:
cit.annotated_regions = []
cit.annotated_regions.append(
TextSpanRegion(start_index=citation.start_block_index, end_index=citation.end_block_index)
)
case _:
logger.debug(f"Unknown citation type encountered: {citation.type}")
annotations.append(cit)
return annotations or None
def service_url(self) -> str:
"""Get the service URL for the chat client.
Returns:
The service URL for the chat client, or None if not set.
"""
return str(self.anthropic_client.base_url)
+88
View File
@@ -0,0 +1,88 @@
[project]
name = "agent-framework-anthropic"
description = "Anthropic integration for Microsoft Agent Framework."
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
readme = "README.md"
requires-python = ">=3.10"
version = "1.0.0b251104"
license-files = ["LICENSE"]
urls.homepage = "https://aka.ms/agent-framework"
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
urls.release_notes = "https://github.com/microsoft/agent-framework/releases?q=tag%3Apython-1&expanded=true"
urls.issues = "https://github.com/microsoft/agent-framework/issues"
classifiers = [
"License :: OSI Approved :: MIT License",
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Typing :: Typed",
]
dependencies = [
"agent-framework-core",
"anthropic>=0.70.0,<1",
]
[tool.uv]
prerelease = "if-necessary-or-explicit"
environments = [
"sys_platform == 'darwin'",
"sys_platform == 'linux'",
"sys_platform == 'win32'"
]
[tool.uv-dynamic-versioning]
fallback-version = "0.0.0"
[tool.pytest.ini_options]
testpaths = 'tests'
addopts = "-ra -q -r fEX"
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
filterwarnings = [
"ignore:Support for class-based `config` is deprecated:DeprecationWarning:pydantic.*"
]
timeout = 120
[tool.ruff]
extend = "../../pyproject.toml"
[tool.coverage.run]
omit = [
"**/__init__.py"
]
[tool.pyright]
extends = "../../pyproject.toml"
exclude = ['tests']
[tool.mypy]
plugins = ['pydantic.mypy']
strict = true
python_version = "3.10"
ignore_missing_imports = true
disallow_untyped_defs = true
no_implicit_optional = true
check_untyped_defs = true
warn_return_any = true
show_error_codes = true
warn_unused_ignores = false
disallow_incomplete_defs = true
disallow_untyped_decorators = true
[tool.bandit]
targets = ["agent_framework_anthropic"]
exclude_dirs = ["tests"]
[tool.poe]
executor.type = "uv"
include = "../../shared_tasks.toml"
[tool.poe.tasks]
mypy = "mypy --config-file $POE_ROOT/pyproject.toml agent_framework_anthropic"
test = "pytest --cov=agent_framework_anthropic --cov-report=term-missing:skip-covered tests"
[build-system]
requires = ["flit-core >= 3.11,<4.0"]
build-backend = "flit_core.buildapi"
@@ -0,0 +1,56 @@
# Copyright (c) Microsoft. All rights reserved.
from typing import Any
from unittest.mock import AsyncMock, MagicMock
from pytest import fixture
@fixture
def exclude_list(request: Any) -> list[str]:
"""Fixture that returns a list of environment variables to exclude."""
return request.param if hasattr(request, "param") else []
@fixture
def override_env_param_dict(request: Any) -> dict[str, str]:
"""Fixture that returns a dict of environment variables to override."""
return request.param if hasattr(request, "param") else {}
@fixture
def anthropic_unit_test_env(monkeypatch, exclude_list, override_env_param_dict): # type: ignore
"""Fixture to set environment variables for AnthropicSettings."""
if exclude_list is None:
exclude_list = []
if override_env_param_dict is None:
override_env_param_dict = {}
env_vars = {
"ANTHROPIC_API_KEY": "test-api-key-12345",
"ANTHROPIC_CHAT_MODEL_ID": "claude-3-5-sonnet-20241022",
}
env_vars.update(override_env_param_dict) # type: ignore
for key, value in env_vars.items():
if key in exclude_list:
monkeypatch.delenv(key, raising=False) # type: ignore
continue
monkeypatch.setenv(key, value) # type: ignore
return env_vars
@fixture
def mock_anthropic_client() -> MagicMock:
"""Fixture that provides a mock AsyncAnthropic client."""
mock_client = MagicMock()
mock_client.base_url = "https://api.anthropic.com"
# Mock beta.messages property
mock_client.beta = MagicMock()
mock_client.beta.messages = MagicMock()
mock_client.beta.messages.create = AsyncMock()
return mock_client
@@ -0,0 +1,777 @@
# Copyright (c) Microsoft. All rights reserved.
import os
from typing import Annotated
from unittest.mock import MagicMock, patch
import pytest
from agent_framework import (
ChatClientProtocol,
ChatMessage,
ChatOptions,
ChatResponseUpdate,
FinishReason,
FunctionCallContent,
FunctionResultContent,
HostedCodeInterpreterTool,
HostedMCPTool,
HostedWebSearchTool,
Role,
TextContent,
TextReasoningContent,
ai_function,
)
from agent_framework.exceptions import ServiceInitializationError
from anthropic.types.beta import (
BetaMessage,
BetaTextBlock,
BetaToolUseBlock,
BetaUsage,
)
from pydantic import Field, ValidationError
from agent_framework_anthropic import AnthropicClient
from agent_framework_anthropic._chat_client import AnthropicSettings
skip_if_anthropic_integration_tests_disabled = pytest.mark.skipif(
os.getenv("RUN_INTEGRATION_TESTS", "false").lower() != "true"
or os.getenv("ANTHROPIC_API_KEY", "") in ("", "test-api-key-12345"),
reason="No real ANTHROPIC_API_KEY provided; skipping integration tests."
if os.getenv("RUN_INTEGRATION_TESTS", "false").lower() == "true"
else "Integration tests are disabled.",
)
def create_test_anthropic_client(
mock_anthropic_client: MagicMock,
model_id: str | None = None,
anthropic_settings: AnthropicSettings | None = None,
) -> AnthropicClient:
"""Helper function to create AnthropicClient instances for testing, bypassing normal validation."""
if anthropic_settings is None:
anthropic_settings = AnthropicSettings(api_key="test-api-key-12345", chat_model_id="claude-3-5-sonnet-20241022")
# Create client instance directly
client = object.__new__(AnthropicClient)
# Set attributes directly
client.anthropic_client = mock_anthropic_client
client.model_id = model_id or anthropic_settings.chat_model_id
client._last_call_id_name = None
client.additional_properties = {}
client.middleware = None
return client
# Settings Tests
def test_anthropic_settings_init(anthropic_unit_test_env: dict[str, str]) -> None:
"""Test AnthropicSettings initialization."""
settings = AnthropicSettings()
assert settings.api_key is not None
assert settings.api_key.get_secret_value() == anthropic_unit_test_env["ANTHROPIC_API_KEY"]
assert settings.chat_model_id == anthropic_unit_test_env["ANTHROPIC_CHAT_MODEL_ID"]
def test_anthropic_settings_init_with_explicit_values() -> None:
"""Test AnthropicSettings initialization with explicit values."""
settings = AnthropicSettings(
api_key="custom-api-key",
chat_model_id="claude-3-opus-20240229",
)
assert settings.api_key is not None
assert settings.api_key.get_secret_value() == "custom-api-key"
assert settings.chat_model_id == "claude-3-opus-20240229"
@pytest.mark.parametrize("exclude_list", [["ANTHROPIC_API_KEY"]], indirect=True)
def test_anthropic_settings_missing_api_key(anthropic_unit_test_env: dict[str, str]) -> None:
"""Test AnthropicSettings when API key is missing."""
settings = AnthropicSettings()
assert settings.api_key is None
assert settings.chat_model_id == anthropic_unit_test_env["ANTHROPIC_CHAT_MODEL_ID"]
# Client Initialization Tests
def test_anthropic_client_init_with_client(mock_anthropic_client: MagicMock) -> None:
"""Test AnthropicClient initialization with existing anthropic_client."""
chat_client = create_test_anthropic_client(mock_anthropic_client, model_id="claude-3-5-sonnet-20241022")
assert chat_client.anthropic_client is mock_anthropic_client
assert chat_client.model_id == "claude-3-5-sonnet-20241022"
assert isinstance(chat_client, ChatClientProtocol)
def test_anthropic_client_init_auto_create_client(anthropic_unit_test_env: dict[str, str]) -> None:
"""Test AnthropicClient initialization with auto-created anthropic_client."""
client = AnthropicClient(
api_key=anthropic_unit_test_env["ANTHROPIC_API_KEY"],
model_id=anthropic_unit_test_env["ANTHROPIC_CHAT_MODEL_ID"],
)
assert client.anthropic_client is not None
assert client.model_id == anthropic_unit_test_env["ANTHROPIC_CHAT_MODEL_ID"]
def test_anthropic_client_init_missing_api_key() -> None:
"""Test AnthropicClient initialization when API key is missing."""
with patch("agent_framework_anthropic._chat_client.AnthropicSettings") as mock_settings:
mock_settings.return_value.api_key = None
mock_settings.return_value.chat_model_id = "claude-3-5-sonnet-20241022"
with pytest.raises(ServiceInitializationError, match="Anthropic API key is required"):
AnthropicClient()
def test_anthropic_client_init_validation_error() -> None:
"""Test that ValidationError in AnthropicSettings is properly handled."""
with patch("agent_framework_anthropic._chat_client.AnthropicSettings") as mock_settings:
mock_settings.side_effect = ValidationError.from_exception_data("test", [])
with pytest.raises(ServiceInitializationError, match="Failed to create Anthropic settings"):
AnthropicClient()
def test_anthropic_client_service_url(mock_anthropic_client: MagicMock) -> None:
"""Test service_url method."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
assert chat_client.service_url() == "https://api.anthropic.com"
# Message Conversion Tests
def test_convert_message_to_anthropic_format_text(mock_anthropic_client: MagicMock) -> None:
"""Test converting text message to Anthropic format."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
message = ChatMessage(role=Role.USER, text="Hello, world!")
result = chat_client._convert_message_to_anthropic_format(message)
assert result["role"] == "user"
assert len(result["content"]) == 1
assert result["content"][0]["type"] == "text"
assert result["content"][0]["text"] == "Hello, world!"
def test_convert_message_to_anthropic_format_function_call(mock_anthropic_client: MagicMock) -> None:
"""Test converting function call message to Anthropic format."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
message = ChatMessage(
role=Role.ASSISTANT,
contents=[
FunctionCallContent(
call_id="call_123",
name="get_weather",
arguments={"location": "San Francisco"},
)
],
)
result = chat_client._convert_message_to_anthropic_format(message)
assert result["role"] == "assistant"
assert len(result["content"]) == 1
assert result["content"][0]["type"] == "tool_use"
assert result["content"][0]["id"] == "call_123"
assert result["content"][0]["name"] == "get_weather"
assert result["content"][0]["input"] == {"location": "San Francisco"}
def test_convert_message_to_anthropic_format_function_result(mock_anthropic_client: MagicMock) -> None:
"""Test converting function result message to Anthropic format."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
message = ChatMessage(
role=Role.TOOL,
contents=[
FunctionResultContent(
call_id="call_123",
name="get_weather",
result="Sunny, 72°F",
)
],
)
result = chat_client._convert_message_to_anthropic_format(message)
assert result["role"] == "user"
assert len(result["content"]) == 1
assert result["content"][0]["type"] == "tool_result"
assert result["content"][0]["tool_use_id"] == "call_123"
# The degree symbol might be escaped differently depending on JSON encoder
assert "Sunny" in result["content"][0]["content"]
assert "72" in result["content"][0]["content"]
assert result["content"][0]["is_error"] is False
def test_convert_message_to_anthropic_format_text_reasoning(mock_anthropic_client: MagicMock) -> None:
"""Test converting text reasoning message to Anthropic format."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
message = ChatMessage(
role=Role.ASSISTANT,
contents=[TextReasoningContent(text="Let me think about this...")],
)
result = chat_client._convert_message_to_anthropic_format(message)
assert result["role"] == "assistant"
assert len(result["content"]) == 1
assert result["content"][0]["type"] == "thinking"
assert result["content"][0]["thinking"] == "Let me think about this..."
def test_convert_messages_to_anthropic_format_with_system(mock_anthropic_client: MagicMock) -> None:
"""Test converting messages list with system message."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
messages = [
ChatMessage(role=Role.SYSTEM, text="You are a helpful assistant."),
ChatMessage(role=Role.USER, text="Hello!"),
]
result = chat_client._convert_messages_to_anthropic_format(messages)
# System message should be skipped
assert len(result) == 1
assert result[0]["role"] == "user"
assert result[0]["content"][0]["text"] == "Hello!"
def test_convert_messages_to_anthropic_format_without_system(mock_anthropic_client: MagicMock) -> None:
"""Test converting messages list without system message."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
messages = [
ChatMessage(role=Role.USER, text="Hello!"),
ChatMessage(role=Role.ASSISTANT, text="Hi there!"),
]
result = chat_client._convert_messages_to_anthropic_format(messages)
assert len(result) == 2
assert result[0]["role"] == "user"
assert result[1]["role"] == "assistant"
# Tool Conversion Tests
def test_convert_tools_to_anthropic_format_ai_function(mock_anthropic_client: MagicMock) -> None:
"""Test converting AIFunction to Anthropic format."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
@ai_function
def get_weather(location: Annotated[str, Field(description="Location to get weather for")]) -> str:
"""Get weather for a location."""
return f"Weather for {location}"
tools = [get_weather]
result = chat_client._convert_tools_to_anthropic_format(tools)
assert result is not None
assert "tools" in result
assert len(result["tools"]) == 1
assert result["tools"][0]["type"] == "custom"
assert result["tools"][0]["name"] == "get_weather"
assert "Get weather for a location" in result["tools"][0]["description"]
def test_convert_tools_to_anthropic_format_web_search(mock_anthropic_client: MagicMock) -> None:
"""Test converting HostedWebSearchTool to Anthropic format."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
tools = [HostedWebSearchTool()]
result = chat_client._convert_tools_to_anthropic_format(tools)
assert result is not None
assert "tools" in result
assert len(result["tools"]) == 1
assert result["tools"][0]["type"] == "web_search_20250305"
assert result["tools"][0]["name"] == "web_search"
def test_convert_tools_to_anthropic_format_code_interpreter(mock_anthropic_client: MagicMock) -> None:
"""Test converting HostedCodeInterpreterTool to Anthropic format."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
tools = [HostedCodeInterpreterTool()]
result = chat_client._convert_tools_to_anthropic_format(tools)
assert result is not None
assert "tools" in result
assert len(result["tools"]) == 1
assert result["tools"][0]["type"] == "code_execution_20250825"
assert result["tools"][0]["name"] == "code_interpreter"
def test_convert_tools_to_anthropic_format_mcp_tool(mock_anthropic_client: MagicMock) -> None:
"""Test converting HostedMCPTool to Anthropic format."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
tools = [HostedMCPTool(name="test-mcp", url="https://example.com/mcp")]
result = chat_client._convert_tools_to_anthropic_format(tools)
assert result is not None
assert "mcp_servers" in result
assert len(result["mcp_servers"]) == 1
assert result["mcp_servers"][0]["type"] == "url"
assert result["mcp_servers"][0]["name"] == "test-mcp"
assert result["mcp_servers"][0]["url"] == "https://example.com/mcp"
def test_convert_tools_to_anthropic_format_mcp_with_auth(mock_anthropic_client: MagicMock) -> None:
"""Test converting HostedMCPTool with authorization headers."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
tools = [
HostedMCPTool(
name="test-mcp",
url="https://example.com/mcp",
headers={"authorization": "Bearer token123"},
)
]
result = chat_client._convert_tools_to_anthropic_format(tools)
assert result is not None
assert "mcp_servers" in result
# The authorization header is converted to authorization_token
assert "authorization_token" in result["mcp_servers"][0]
assert result["mcp_servers"][0]["authorization_token"] == "Bearer token123"
def test_convert_tools_to_anthropic_format_dict_tool(mock_anthropic_client: MagicMock) -> None:
"""Test converting dict tool to Anthropic format."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
tools = [{"type": "custom", "name": "custom_tool", "description": "A custom tool"}]
result = chat_client._convert_tools_to_anthropic_format(tools)
assert result is not None
assert "tools" in result
assert len(result["tools"]) == 1
assert result["tools"][0]["name"] == "custom_tool"
def test_convert_tools_to_anthropic_format_none(mock_anthropic_client: MagicMock) -> None:
"""Test converting None tools."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
result = chat_client._convert_tools_to_anthropic_format(None)
assert result is None
# Run Options Tests
async def test_create_run_options_basic(mock_anthropic_client: MagicMock) -> None:
"""Test _create_run_options with basic ChatOptions."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
messages = [ChatMessage(role=Role.USER, text="Hello")]
chat_options = ChatOptions(max_tokens=100, temperature=0.7)
run_options = chat_client._create_run_options(messages, chat_options)
assert run_options["model"] == chat_client.model_id
assert run_options["max_tokens"] == 100
assert run_options["temperature"] == 0.7
assert "messages" in run_options
async def test_create_run_options_with_system_message(mock_anthropic_client: MagicMock) -> None:
"""Test _create_run_options with system message."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
messages = [
ChatMessage(role=Role.SYSTEM, text="You are helpful."),
ChatMessage(role=Role.USER, text="Hello"),
]
chat_options = ChatOptions()
run_options = chat_client._create_run_options(messages, chat_options)
assert run_options["system"] == "You are helpful."
assert len(run_options["messages"]) == 1 # System message not in messages list
async def test_create_run_options_with_tool_choice_auto(mock_anthropic_client: MagicMock) -> None:
"""Test _create_run_options with auto tool choice."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
messages = [ChatMessage(role=Role.USER, text="Hello")]
chat_options = ChatOptions(tool_choice="auto")
run_options = chat_client._create_run_options(messages, chat_options)
assert run_options["tool_choice"]["type"] == "auto"
async def test_create_run_options_with_tool_choice_required(mock_anthropic_client: MagicMock) -> None:
"""Test _create_run_options with required tool choice."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
messages = [ChatMessage(role=Role.USER, text="Hello")]
# For required with specific function, need to pass as dict
chat_options = ChatOptions(tool_choice={"mode": "required", "required_function_name": "get_weather"})
run_options = chat_client._create_run_options(messages, chat_options)
assert run_options["tool_choice"]["type"] == "tool"
assert run_options["tool_choice"]["name"] == "get_weather"
async def test_create_run_options_with_tool_choice_none(mock_anthropic_client: MagicMock) -> None:
"""Test _create_run_options with none tool choice."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
messages = [ChatMessage(role=Role.USER, text="Hello")]
chat_options = ChatOptions(tool_choice="none")
run_options = chat_client._create_run_options(messages, chat_options)
assert run_options["tool_choice"]["type"] == "none"
async def test_create_run_options_with_tools(mock_anthropic_client: MagicMock) -> None:
"""Test _create_run_options with tools."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
@ai_function
def get_weather(location: str) -> str:
"""Get weather for a location."""
return f"Weather for {location}"
messages = [ChatMessage(role=Role.USER, text="Hello")]
chat_options = ChatOptions(tools=[get_weather])
run_options = chat_client._create_run_options(messages, chat_options)
assert "tools" in run_options
assert len(run_options["tools"]) == 1
async def test_create_run_options_with_stop_sequences(mock_anthropic_client: MagicMock) -> None:
"""Test _create_run_options with stop sequences."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
messages = [ChatMessage(role=Role.USER, text="Hello")]
chat_options = ChatOptions(stop=["STOP", "END"])
run_options = chat_client._create_run_options(messages, chat_options)
assert run_options["stop_sequences"] == ["STOP", "END"]
async def test_create_run_options_with_top_p(mock_anthropic_client: MagicMock) -> None:
"""Test _create_run_options with top_p."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
messages = [ChatMessage(role=Role.USER, text="Hello")]
chat_options = ChatOptions(top_p=0.9)
run_options = chat_client._create_run_options(messages, chat_options)
assert run_options["top_p"] == 0.9
# Response Processing Tests
def test_process_message_basic(mock_anthropic_client: MagicMock) -> None:
"""Test _process_message with basic text response."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
mock_message = MagicMock(spec=BetaMessage)
mock_message.id = "msg_123"
mock_message.model = "claude-3-5-sonnet-20241022"
mock_message.content = [BetaTextBlock(type="text", text="Hello there!")]
mock_message.usage = BetaUsage(input_tokens=10, output_tokens=5)
mock_message.stop_reason = "end_turn"
response = chat_client._process_message(mock_message)
assert response.response_id == "msg_123"
assert response.model_id == "claude-3-5-sonnet-20241022"
assert len(response.messages) == 1
assert response.messages[0].role == Role.ASSISTANT
assert len(response.messages[0].contents) == 1
assert isinstance(response.messages[0].contents[0], TextContent)
assert response.messages[0].contents[0].text == "Hello there!"
assert response.finish_reason == FinishReason.STOP
assert response.usage_details is not None
assert response.usage_details.input_token_count == 10
assert response.usage_details.output_token_count == 5
def test_process_message_with_tool_use(mock_anthropic_client: MagicMock) -> None:
"""Test _process_message with tool use."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
mock_message = MagicMock(spec=BetaMessage)
mock_message.id = "msg_123"
mock_message.model = "claude-3-5-sonnet-20241022"
mock_message.content = [
BetaToolUseBlock(
type="tool_use",
id="call_123",
name="get_weather",
input={"location": "San Francisco"},
)
]
mock_message.usage = BetaUsage(input_tokens=10, output_tokens=5)
mock_message.stop_reason = "tool_use"
response = chat_client._process_message(mock_message)
assert len(response.messages[0].contents) == 1
assert isinstance(response.messages[0].contents[0], FunctionCallContent)
assert response.messages[0].contents[0].call_id == "call_123"
assert response.messages[0].contents[0].name == "get_weather"
assert response.finish_reason == FinishReason.TOOL_CALLS
def test_parse_message_usage_basic(mock_anthropic_client: MagicMock) -> None:
"""Test _parse_message_usage with basic usage."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
usage = BetaUsage(input_tokens=10, output_tokens=5)
result = chat_client._parse_message_usage(usage)
assert result is not None
assert result.input_token_count == 10
assert result.output_token_count == 5
def test_parse_message_usage_none(mock_anthropic_client: MagicMock) -> None:
"""Test _parse_message_usage with None usage."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
result = chat_client._parse_message_usage(None)
assert result is None
def test_parse_message_contents_text(mock_anthropic_client: MagicMock) -> None:
"""Test _parse_message_contents with text content."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
content = [BetaTextBlock(type="text", text="Hello!")]
result = chat_client._parse_message_contents(content)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "Hello!"
def test_parse_message_contents_tool_use(mock_anthropic_client: MagicMock) -> None:
"""Test _parse_message_contents with tool use."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
content = [
BetaToolUseBlock(
type="tool_use",
id="call_123",
name="get_weather",
input={"location": "SF"},
)
]
result = chat_client._parse_message_contents(content)
assert len(result) == 1
assert isinstance(result[0], FunctionCallContent)
assert result[0].call_id == "call_123"
assert result[0].name == "get_weather"
# Stream Processing Tests
def test_process_stream_event_simple(mock_anthropic_client: MagicMock) -> None:
"""Test _process_stream_event with simple mock event."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
# Test with a basic mock event - the actual implementation will handle real events
mock_event = MagicMock()
mock_event.type = "message_stop"
result = chat_client._process_stream_event(mock_event)
# message_stop events return None
assert result is None
async def test_inner_get_response(mock_anthropic_client: MagicMock) -> None:
"""Test _inner_get_response method."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
# Create a mock message response
mock_message = MagicMock(spec=BetaMessage)
mock_message.id = "msg_test"
mock_message.model = "claude-3-5-sonnet-20241022"
mock_message.content = [BetaTextBlock(type="text", text="Hello!")]
mock_message.usage = BetaUsage(input_tokens=5, output_tokens=3)
mock_message.stop_reason = "end_turn"
mock_anthropic_client.beta.messages.create.return_value = mock_message
messages = [ChatMessage(role=Role.USER, text="Hi")]
chat_options = ChatOptions(max_tokens=10)
response = await chat_client._inner_get_response( # type: ignore[attr-defined]
messages=messages, chat_options=chat_options
)
assert response is not None
assert response.response_id == "msg_test"
assert len(response.messages) == 1
async def test_inner_get_streaming_response(mock_anthropic_client: MagicMock) -> None:
"""Test _inner_get_streaming_response method."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
# Create mock streaming response
async def mock_stream():
mock_event = MagicMock()
mock_event.type = "message_stop"
yield mock_event
mock_anthropic_client.beta.messages.create.return_value = mock_stream()
messages = [ChatMessage(role=Role.USER, text="Hi")]
chat_options = ChatOptions(max_tokens=10)
chunks: list[ChatResponseUpdate] = []
async for chunk in chat_client._inner_get_streaming_response( # type: ignore[attr-defined]
messages=messages, chat_options=chat_options
):
if chunk:
chunks.append(chunk)
# We should get at least some response (even if empty due to message_stop)
assert isinstance(chunks, list)
# Integration Tests
@ai_function
def get_weather(
location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
"""Get the weather for a location."""
return f"The weather in {location} is sunny and 72°F"
@pytest.mark.flaky
@skip_if_anthropic_integration_tests_disabled
async def test_anthropic_client_integration_basic_chat() -> None:
"""Integration test for basic chat completion."""
client = AnthropicClient()
messages = [ChatMessage(role=Role.USER, text="Say 'Hello, World!' and nothing else.")]
response = await client.get_response(messages=messages, chat_options=ChatOptions(max_tokens=50))
assert response is not None
assert len(response.messages) > 0
assert response.messages[0].role == Role.ASSISTANT
assert len(response.messages[0].text) > 0
assert response.usage_details is not None
@pytest.mark.flaky
@skip_if_anthropic_integration_tests_disabled
async def test_anthropic_client_integration_streaming_chat() -> None:
"""Integration test for streaming chat completion."""
client = AnthropicClient()
messages = [ChatMessage(role=Role.USER, text="Count from 1 to 5.")]
chunks = []
async for chunk in client.get_streaming_response(messages=messages, chat_options=ChatOptions(max_tokens=50)):
chunks.append(chunk)
assert len(chunks) > 0
assert any(chunk.contents for chunk in chunks)
@pytest.mark.flaky
@skip_if_anthropic_integration_tests_disabled
async def test_anthropic_client_integration_function_calling() -> None:
"""Integration test for function calling."""
client = AnthropicClient()
messages = [ChatMessage(role=Role.USER, text="What's the weather in San Francisco?")]
tools = [get_weather]
response = await client.get_response(
messages=messages,
chat_options=ChatOptions(tools=tools, max_tokens=100),
)
assert response is not None
# Should contain function call
has_function_call = any(
isinstance(content, FunctionCallContent) for msg in response.messages for content in msg.contents
)
assert has_function_call
@pytest.mark.flaky
@skip_if_anthropic_integration_tests_disabled
async def test_anthropic_client_integration_with_system_message() -> None:
"""Integration test with system message."""
client = AnthropicClient()
messages = [
ChatMessage(role=Role.SYSTEM, text="You are a pirate. Always respond like a pirate."),
ChatMessage(role=Role.USER, text="Hello!"),
]
response = await client.get_response(messages=messages, chat_options=ChatOptions(max_tokens=50))
assert response is not None
assert len(response.messages) > 0
@pytest.mark.flaky
@skip_if_anthropic_integration_tests_disabled
async def test_anthropic_client_integration_temperature_control() -> None:
"""Integration test with temperature control."""
client = AnthropicClient()
messages = [ChatMessage(role=Role.USER, text="Say hello.")]
response = await client.get_response(
messages=messages,
chat_options=ChatOptions(max_tokens=20, temperature=0.0),
)
assert response is not None
assert response.messages[0].text is not None
@pytest.mark.flaky
@skip_if_anthropic_integration_tests_disabled
async def test_anthropic_client_integration_ordering() -> None:
"""Integration test with ordering."""
client = AnthropicClient()
messages = [
ChatMessage(role=Role.USER, text="Say hello."),
ChatMessage(role=Role.USER, text="Then say goodbye."),
ChatMessage(role=Role.ASSISTANT, text="Thank you for chatting!"),
ChatMessage(role=Role.ASSISTANT, text="Let me know if I can help."),
ChatMessage(role=Role.USER, text="Just testing things."),
]
response = await client.get_response(messages=messages)
assert response is not None
assert response.messages[0].text is not None
@@ -735,10 +735,10 @@ class AzureAIAgentClient(BaseChatClient):
chat_tool_mode = chat_options.tool_choice
if chat_tool_mode is None or chat_tool_mode == ToolMode.NONE or chat_tool_mode == "none":
chat_options.tools = None
chat_options.tool_choice = ToolMode.NONE.mode
chat_options.tool_choice = ToolMode.NONE
return
chat_options.tool_choice = chat_tool_mode.mode if isinstance(chat_tool_mode, ToolMode) else chat_tool_mode
chat_options.tool_choice = chat_tool_mode
async def _create_run_options(
self,
+1 -1
View File
@@ -4,7 +4,7 @@ description = "Azure AI Foundry integration for Microsoft Agent Framework."
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
readme = "README.md"
requires-python = ">=3.10"
version = "1.0.0b251028"
version = "1.0.0b251104"
license-files = ["LICENSE"]
urls.homepage = "https://aka.ms/agent-framework"
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -559,19 +559,6 @@ async def test_azure_ai_chat_client_create_run_options_with_messages(mock_ai_pro
assert len(run_options["additional_messages"]) == 1 # Only user message
async def test_azure_ai_chat_client_instructions_sent_once(mock_ai_project_client: MagicMock) -> None:
"""Ensure instructions are only sent once for AzureAIAgentClient."""
chat_client = create_test_azure_ai_chat_client(mock_ai_project_client)
instructions = "You are a helpful assistant."
chat_options = ChatOptions(instructions=instructions)
messages = chat_client.prepare_messages([ChatMessage(role=Role.USER, text="Hello")], chat_options)
run_options, _ = await chat_client._create_run_options(messages, chat_options) # type: ignore
assert run_options.get("instructions") == instructions
async def test_azure_ai_chat_client_inner_get_response(mock_ai_project_client: MagicMock) -> None:
"""Test _inner_get_response method."""
chat_client = create_test_azure_ai_chat_client(mock_ai_project_client, agent_id="test-agent")
+1 -1
View File
@@ -4,7 +4,7 @@ description = "Copilot Studio integration for Microsoft Agent Framework."
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
readme = "README.md"
requires-python = ">=3.10"
version = "1.0.0b251028"
version = "1.0.0b251104"
license-files = ["LICENSE"]
urls.homepage = "https://aka.ms/agent-framework"
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -20,13 +20,7 @@ from ._middleware import (
from ._serialization import SerializationMixin
from ._threads import ChatMessageStoreProtocol
from ._tools import ToolProtocol
from ._types import (
ChatMessage,
ChatOptions,
ChatResponse,
ChatResponseUpdate,
ToolMode,
)
from ._types import ChatMessage, ChatOptions, ChatResponse, ChatResponseUpdate, ToolMode, prepare_messages
if TYPE_CHECKING:
from ._agents import ChatAgent
@@ -216,28 +210,7 @@ class ChatClientProtocol(Protocol):
# region ChatClientBase
def prepare_messages(messages: str | ChatMessage | list[str] | list[ChatMessage]) -> list[ChatMessage]:
"""Convert various message input formats into a list of ChatMessage objects.
Args:
messages: The input messages in various supported formats.
Returns:
A list of ChatMessage objects.
"""
if isinstance(messages, str):
return [ChatMessage(role="user", text=messages)]
if isinstance(messages, ChatMessage):
return [messages]
return_messages: list[ChatMessage] = []
for msg in messages:
if isinstance(msg, str):
msg = ChatMessage(role="user", text=msg)
return_messages.append(msg)
return return_messages
def merge_chat_options(
def _merge_chat_options(
*,
base_chat_options: ChatOptions | Any | None,
model_id: str | None = None,
@@ -405,25 +378,6 @@ class BaseChatClient(SerializationMixin, ABC):
return result
def prepare_messages(
self, messages: str | ChatMessage | list[str] | list[ChatMessage], chat_options: ChatOptions
) -> MutableSequence[ChatMessage]:
"""Convert various message input formats into a list of ChatMessage objects.
Prepends system instructions if present in chat_options.
Args:
messages: The input messages in various supported formats.
chat_options: The chat options containing instructions and other settings.
Returns:
A mutable sequence of ChatMessage objects.
"""
if chat_options.instructions:
system_msg = ChatMessage(role="system", text=chat_options.instructions)
return [system_msg, *prepare_messages(messages)]
return prepare_messages(messages)
def _filter_internal_kwargs(self, kwargs: dict[str, Any]) -> dict[str, Any]:
"""Filter out internal framework parameters that shouldn't be passed to chat client implementations.
@@ -584,7 +538,7 @@ class BaseChatClient(SerializationMixin, ABC):
"""
# Normalize tools and merge with base chat_options
normalized_tools = await self._normalize_tools(tools)
chat_options = merge_chat_options(
chat_options = _merge_chat_options(
base_chat_options=kwargs.pop("chat_options", None),
model_id=model_id,
frequency_penalty=frequency_penalty,
@@ -612,7 +566,11 @@ class BaseChatClient(SerializationMixin, ABC):
)
chat_options.store = True
prepped_messages = self.prepare_messages(messages, chat_options)
if chat_options.instructions:
system_msg = ChatMessage(role="system", text=chat_options.instructions)
prepped_messages = [system_msg, *prepare_messages(messages)]
else:
prepped_messages = prepare_messages(messages)
self._prepare_tool_choice(chat_options=chat_options)
filtered_kwargs = self._filter_internal_kwargs(kwargs)
@@ -679,7 +637,7 @@ class BaseChatClient(SerializationMixin, ABC):
"""
# Normalize tools and merge with base chat_options
normalized_tools = await self._normalize_tools(tools)
chat_options = merge_chat_options(
chat_options = _merge_chat_options(
base_chat_options=kwargs.pop("chat_options", None),
model_id=model_id,
frequency_penalty=frequency_penalty,
@@ -707,7 +665,11 @@ class BaseChatClient(SerializationMixin, ABC):
)
chat_options.store = True
prepped_messages = self.prepare_messages(messages, chat_options)
if chat_options.instructions:
system_msg = ChatMessage(role="system", text=chat_options.instructions)
prepped_messages = [system_msg, *prepare_messages(messages)]
else:
prepped_messages = prepare_messages(messages)
self._prepare_tool_choice(chat_options=chat_options)
filtered_kwargs = self._filter_internal_kwargs(kwargs)
@@ -728,12 +690,12 @@ class BaseChatClient(SerializationMixin, ABC):
chat_tool_mode = chat_options.tool_choice
if chat_tool_mode is None or chat_tool_mode == ToolMode.NONE or chat_tool_mode == "none":
chat_options.tools = None
chat_options.tool_choice = ToolMode.NONE.mode
chat_options.tool_choice = ToolMode.NONE
return
if not chat_options.tools:
chat_options.tool_choice = ToolMode.NONE.mode
chat_options.tool_choice = ToolMode.NONE
else:
chat_options.tool_choice = chat_tool_mode.mode if isinstance(chat_tool_mode, ToolMode) else chat_tool_mode
chat_options.tool_choice = chat_tool_mode
def service_url(self) -> str:
"""Get the URL of the service.
@@ -8,7 +8,7 @@ from functools import update_wrapper
from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypeAlias, TypeVar
from ._serialization import SerializationMixin
from ._types import AgentRunResponse, AgentRunResponseUpdate, ChatMessage
from ._types import AgentRunResponse, AgentRunResponseUpdate, ChatMessage, prepare_messages
from .exceptions import MiddlewareException
if TYPE_CHECKING:
@@ -1375,7 +1375,7 @@ def use_chat_middleware(chat_client_class: type[TChatClient]) -> type[TChatClien
pipeline = ChatMiddlewarePipeline(chat_middleware_list) # type: ignore[arg-type]
context = ChatContext(
chat_client=self,
messages=self.prepare_messages(messages, chat_options),
messages=prepare_messages(messages),
chat_options=chat_options,
is_streaming=False,
kwargs=kwargs,
@@ -1425,7 +1425,7 @@ def use_chat_middleware(chat_client_class: type[TChatClient]) -> type[TChatClien
pipeline = ChatMiddlewarePipeline(all_middleware) # type: ignore[arg-type]
context = ChatContext(
chat_client=self,
messages=self.prepare_messages(messages, chat_options),
messages=prepare_messages(messages),
chat_options=chat_options,
is_streaming=True,
kwargs=kwargs,
@@ -1318,13 +1318,13 @@ def _handle_function_calls_response(
messages: "str | ChatMessage | list[str] | list[ChatMessage]",
**kwargs: Any,
) -> "ChatResponse":
from ._clients import prepare_messages
from ._middleware import extract_and_merge_function_middleware
from ._types import (
ChatMessage,
FunctionApprovalRequestContent,
FunctionCallContent,
FunctionResultContent,
prepare_messages,
)
# Extract and merge function middleware from chat client with kwargs pipeline
@@ -1465,7 +1465,6 @@ def _handle_function_calls_streaming_response(
**kwargs: Any,
) -> AsyncIterable["ChatResponseUpdate"]:
"""Wrap the inner get streaming response method to handle tool calls."""
from ._clients import prepare_messages
from ._middleware import extract_and_merge_function_middleware
from ._types import (
ChatMessage,
@@ -1473,6 +1472,7 @@ def _handle_function_calls_streaming_response(
ChatResponseUpdate,
FunctionCallContent,
FunctionResultContent,
prepare_messages,
)
# Extract and merge function middleware from chat client with kwargs pipeline
+36 -15
View File
@@ -561,7 +561,7 @@ class BaseContent(SerializationMixin):
def __init__(
self,
*,
annotations: list[Annotations | MutableMapping[str, Any]] | None = None,
annotations: Sequence[Annotations | MutableMapping[str, Any]] | None = None,
additional_properties: dict[str, Any] | None = None,
raw_representation: Any | None = None,
**kwargs: Any,
@@ -651,7 +651,7 @@ class TextContent(BaseContent):
*,
additional_properties: dict[str, Any] | None = None,
raw_representation: Any | None = None,
annotations: list[Annotations | MutableMapping[str, Any]] | None = None,
annotations: Sequence[Annotations | MutableMapping[str, Any]] | None = None,
**kwargs: Any,
):
"""Initializes a TextContent instance.
@@ -793,7 +793,7 @@ class TextReasoningContent(BaseContent):
*,
additional_properties: dict[str, Any] | None = None,
raw_representation: Any | None = None,
annotations: list[Annotations | MutableMapping[str, Any]] | None = None,
annotations: Sequence[Annotations | MutableMapping[str, Any]] | None = None,
**kwargs: Any,
):
"""Initializes a TextReasoningContent instance.
@@ -936,7 +936,7 @@ class DataContent(BaseContent):
self,
*,
uri: str,
annotations: list[Annotations | MutableMapping[str, Any]] | None = None,
annotations: Sequence[Annotations | MutableMapping[str, Any]] | None = None,
additional_properties: dict[str, Any] | None = None,
raw_representation: Any | None = None,
**kwargs: Any,
@@ -962,7 +962,7 @@ class DataContent(BaseContent):
*,
data: bytes,
media_type: str,
annotations: list[Annotations | MutableMapping[str, Any]] | None = None,
annotations: Sequence[Annotations | MutableMapping[str, Any]] | None = None,
additional_properties: dict[str, Any] | None = None,
raw_representation: Any | None = None,
**kwargs: Any,
@@ -989,7 +989,7 @@ class DataContent(BaseContent):
uri: str | None = None,
data: bytes | None = None,
media_type: str | None = None,
annotations: list[Annotations | MutableMapping[str, Any]] | None = None,
annotations: Sequence[Annotations | MutableMapping[str, Any]] | None = None,
additional_properties: dict[str, Any] | None = None,
raw_representation: Any | None = None,
**kwargs: Any,
@@ -1093,7 +1093,7 @@ class UriContent(BaseContent):
uri: str,
media_type: str,
*,
annotations: list[Annotations | MutableMapping[str, Any]] | None = None,
annotations: Sequence[Annotations | MutableMapping[str, Any]] | None = None,
additional_properties: dict[str, Any] | None = None,
raw_representation: Any | None = None,
**kwargs: Any,
@@ -1187,7 +1187,7 @@ class ErrorContent(BaseContent):
message: str | None = None,
error_code: str | None = None,
details: str | None = None,
annotations: list[Annotations | MutableMapping[str, Any]] | None = None,
annotations: Sequence[Annotations | MutableMapping[str, Any]] | None = None,
additional_properties: dict[str, Any] | None = None,
raw_representation: Any | None = None,
**kwargs: Any,
@@ -1271,7 +1271,7 @@ class FunctionCallContent(BaseContent):
name: str,
arguments: str | dict[str, Any | None] | None = None,
exception: Exception | None = None,
annotations: list[Annotations | MutableMapping[str, Any]] | None = None,
annotations: Sequence[Annotations | MutableMapping[str, Any]] | None = None,
additional_properties: dict[str, Any] | None = None,
raw_representation: Any | None = None,
**kwargs: Any,
@@ -1380,7 +1380,7 @@ class FunctionResultContent(BaseContent):
call_id: str,
result: Any | None = None,
exception: Exception | None = None,
annotations: list[Annotations | MutableMapping[str, Any]] | None = None,
annotations: Sequence[Annotations | MutableMapping[str, Any]] | None = None,
additional_properties: dict[str, Any] | None = None,
raw_representation: Any | None = None,
**kwargs: Any,
@@ -1438,7 +1438,7 @@ class UsageContent(BaseContent):
self,
details: UsageDetails | MutableMapping[str, Any],
*,
annotations: list[Annotations | MutableMapping[str, Any]] | None = None,
annotations: Sequence[Annotations | MutableMapping[str, Any]] | None = None,
additional_properties: dict[str, Any] | None = None,
raw_representation: Any | None = None,
**kwargs: Any,
@@ -1556,7 +1556,7 @@ class BaseUserInputRequest(BaseContent):
self,
*,
id: str,
annotations: list[Annotations | MutableMapping[str, Any]] | None = None,
annotations: Sequence[Annotations | MutableMapping[str, Any]] | None = None,
additional_properties: dict[str, Any] | None = None,
raw_representation: Any | None = None,
**kwargs: Any,
@@ -1610,7 +1610,7 @@ class FunctionApprovalResponseContent(BaseContent):
*,
id: str,
function_call: FunctionCallContent | MutableMapping[str, Any],
annotations: list[Annotations | MutableMapping[str, Any]] | None = None,
annotations: Sequence[Annotations | MutableMapping[str, Any]] | None = None,
additional_properties: dict[str, Any] | None = None,
raw_representation: Any | None = None,
**kwargs: Any,
@@ -1674,7 +1674,7 @@ class FunctionApprovalRequestContent(BaseContent):
*,
id: str,
function_call: FunctionCallContent | MutableMapping[str, Any],
annotations: list[Annotations | MutableMapping[str, Any]] | None = None,
annotations: Sequence[Annotations | MutableMapping[str, Any]] | None = None,
additional_properties: dict[str, Any] | None = None,
raw_representation: Any | None = None,
**kwargs: Any,
@@ -2052,6 +2052,27 @@ class ChatMessage(SerializationMixin):
return " ".join(content.text for content in self.contents if isinstance(content, TextContent))
def prepare_messages(messages: str | ChatMessage | list[str] | list[ChatMessage]) -> list[ChatMessage]:
"""Convert various message input formats into a list of ChatMessage objects.
Args:
messages: The input messages in various supported formats.
Returns:
A list of ChatMessage objects.
"""
if isinstance(messages, str):
return [ChatMessage(role="user", text=messages)]
if isinstance(messages, ChatMessage):
return [messages]
return_messages: list[ChatMessage] = []
for msg in messages:
if isinstance(msg, str):
msg = ChatMessage(role="user", text=msg)
return_messages.append(msg)
return return_messages
# region ChatResponse
@@ -3125,7 +3146,7 @@ class ChatOptions(SerializationMixin):
@classmethod
def _validate_tool_mode(
cls, tool_choice: ToolMode | Literal["auto", "required", "none"] | Mapping[str, Any] | None
) -> ToolMode | str | None:
) -> ToolMode | None:
"""Validates the tool_choice field to ensure it is a valid ToolMode."""
if not tool_choice:
return None
@@ -60,10 +60,13 @@ class WorkflowAgent(BaseAgent):
@classmethod
def from_json(cls, raw: str) -> "WorkflowAgent.RequestInfoFunctionArgs":
data = json.loads(raw)
if not isinstance(data, dict):
try:
parsed: Any = json.loads(raw)
except json.JSONDecodeError as exc:
raise ValueError(f"RequestInfoFunctionArgs JSON payload is malformed: {exc}") from exc
if not isinstance(parsed, dict):
raise ValueError("RequestInfoFunctionArgs JSON payload must decode to a mapping")
return cls.from_dict(data)
return cls.from_dict(cast(dict[str, Any], parsed))
def __init__(
self,
@@ -80,6 +80,14 @@ def _clone_chat_agent(agent: ChatAgent) -> ChatAgent:
options = agent.chat_options
middleware = list(agent.middleware or [])
# Reconstruct the original tools list by combining regular tools with MCP tools.
# ChatAgent.__init__ separates MCP tools into _local_mcp_tools during initialization,
# so we need to recombine them here to pass the complete tools list to the constructor.
# This makes sure MCP tools are preserved when cloning agents for handoff workflows.
all_tools = list(options.tools) if options.tools else []
if agent._local_mcp_tools:
all_tools.extend(agent._local_mcp_tools)
return ChatAgent(
chat_client=agent.chat_client,
instructions=options.instructions,
@@ -101,7 +109,7 @@ def _clone_chat_agent(agent: ChatAgent) -> ChatAgent:
store=options.store,
temperature=options.temperature,
tool_choice=options.tool_choice, # type: ignore[arg-type]
tools=list(options.tools) if options.tools else None,
tools=all_tools if all_tools else None,
top_p=options.top_p,
user=options.user,
additional_chat_options=dict(options.additional_properties),
@@ -336,10 +344,15 @@ class _HandoffCoordinator(BaseGroupChatOrchestrator):
if await self._check_termination():
logger.info("Handoff workflow termination condition met. Ending conversation.")
await ctx.yield_output(list(conversation))
# Clean the output conversation for display
cleaned_output = clean_conversation_for_handoff(conversation)
await ctx.yield_output(cleaned_output)
return
await ctx.send_message(list(conversation), target_id=self._input_gateway_id)
# Clean conversation before sending to gateway for user input request
# This removes tool messages that shouldn't be shown to users
cleaned_for_display = clean_conversation_for_handoff(conversation)
await ctx.send_message(cleaned_for_display, target_id=self._input_gateway_id)
@handler
async def handle_user_input(
@@ -1274,12 +1287,12 @@ class HandoffBuilder:
updated_executor, tool_targets = self._prepare_agent_with_handoffs(executor, targets_map)
self._executors[source_exec_id] = updated_executor
handoff_tool_targets.update(tool_targets)
else:
# Default behavior: only coordinator gets handoff tools to all specialists
if isinstance(starting_executor, AgentExecutor) and specialists:
starting_executor, tool_targets = self._prepare_agent_with_handoffs(starting_executor, specialists)
self._executors[self._starting_agent_id] = starting_executor
handoff_tool_targets.update(tool_targets) # Update references after potential agent modifications
else:
# Default behavior: only coordinator gets handoff tools to all specialists
if isinstance(starting_executor, AgentExecutor) and specialists:
starting_executor, tool_targets = self._prepare_agent_with_handoffs(starting_executor, specialists)
self._executors[self._starting_agent_id] = starting_executor
handoff_tool_targets.update(tool_targets) # Update references after potential agent modifications
starting_executor = self._executors[self._starting_agent_id]
specialists = {
exec_id: executor for exec_id, executor in self._executors.items() if exec_id != self._starting_agent_id
@@ -2442,16 +2442,6 @@ class MagenticWorkflow:
f"Missing names: {missing}; unexpected names: {unexpected}."
)
async def run_stream_from_checkpoint(
self,
checkpoint_id: str,
checkpoint_storage: CheckpointStorage | None = None,
) -> AsyncIterable[WorkflowEvent]:
"""Resume orchestration from a checkpoint and stream resulting events."""
await self._validate_checkpoint_participants(checkpoint_id, checkpoint_storage)
async for event in self._workflow.run_stream_from_checkpoint(checkpoint_id, checkpoint_storage):
yield event
async def run_with_string(self, task_text: str) -> WorkflowRunResult:
"""Run the workflow with a task string and return all events.
@@ -2495,32 +2485,6 @@ class MagenticWorkflow:
events.append(event)
return WorkflowRunResult(events)
async def run_from_checkpoint(
self,
checkpoint_id: str,
checkpoint_storage: CheckpointStorage | None = None,
) -> WorkflowRunResult:
"""Resume orchestration from a checkpoint and collect all resulting events."""
events: list[WorkflowEvent] = []
async for event in self.run_stream_from_checkpoint(checkpoint_id, checkpoint_storage):
events.append(event)
return WorkflowRunResult(events)
async def send_responses_streaming(self, responses: dict[str, Any]) -> AsyncIterable[WorkflowEvent]:
"""Forward responses to pending requests and stream resulting events.
This delegates to the underlying Workflow implementation.
"""
async for event in self._workflow.send_responses_streaming(responses):
yield event
async def send_responses(self, responses: dict[str, Any]) -> WorkflowRunResult:
"""Forward responses to pending requests and return all resulting events.
This delegates to the underlying Workflow implementation.
"""
return await self._workflow.send_responses(responses)
def __getattr__(self, name: str) -> Any:
"""Delegate unknown attributes to the underlying workflow."""
return getattr(self._workflow, name)
@@ -63,11 +63,12 @@ def clean_conversation_for_handoff(conversation: list[ChatMessage]) -> list[Chat
# Has tool content - only keep if it also has text
if msg.text and msg.text.strip():
# Create fresh text-only message
# Create fresh text-only message while preserving additional_properties
msg_copy = ChatMessage(
role=msg.role,
text=msg.text,
author_name=msg.author_name,
additional_properties=dict(msg.additional_properties) if msg.additional_properties else None,
)
cleaned.append(msg_copy)
@@ -171,6 +171,18 @@ class RunnerContext(Protocol):
"""
...
def set_runtime_checkpoint_storage(self, storage: CheckpointStorage) -> None:
"""Set runtime checkpoint storage to override build-time configuration.
Args:
storage: The checkpoint storage to use for this run.
"""
...
def clear_runtime_checkpoint_storage(self) -> None:
"""Clear runtime checkpoint storage override."""
...
# Checkpointing APIs (optional, enabled by storage)
def set_workflow_id(self, workflow_id: str) -> None:
"""Set the workflow ID for the context."""
@@ -279,6 +291,7 @@ class InProcRunnerContext:
# Checkpointing configuration/state
self._checkpoint_storage = checkpoint_storage
self._runtime_checkpoint_storage: CheckpointStorage | None = None
self._workflow_id: str | None = None
# Streaming flag - set by workflow's run_stream() vs run()
@@ -329,8 +342,28 @@ class InProcRunnerContext:
# region Checkpointing
def _get_effective_checkpoint_storage(self) -> CheckpointStorage | None:
"""Get the effective checkpoint storage (runtime override or build-time)."""
return self._runtime_checkpoint_storage or self._checkpoint_storage
def set_runtime_checkpoint_storage(self, storage: CheckpointStorage) -> None:
"""Set runtime checkpoint storage to override build-time configuration.
Args:
storage: The checkpoint storage to use for this run.
"""
self._runtime_checkpoint_storage = storage
def clear_runtime_checkpoint_storage(self) -> None:
"""Clear runtime checkpoint storage override.
This is called automatically by workflow execution methods after a run completes,
ensuring runtime storage doesn't leak across runs.
"""
self._runtime_checkpoint_storage = None
def has_checkpointing(self) -> bool:
return self._checkpoint_storage is not None
return self._get_effective_checkpoint_storage() is not None
async def create_checkpoint(
self,
@@ -338,7 +371,8 @@ class InProcRunnerContext:
iteration_count: int,
metadata: dict[str, Any] | None = None,
) -> str:
if not self._checkpoint_storage:
storage = self._get_effective_checkpoint_storage()
if not storage:
raise ValueError("Checkpoint storage not configured")
self._workflow_id = self._workflow_id or str(uuid.uuid4())
@@ -352,19 +386,21 @@ class InProcRunnerContext:
iteration_count=state["iteration_count"],
metadata=metadata or {},
)
checkpoint_id = await self._checkpoint_storage.save_checkpoint(checkpoint)
checkpoint_id = await storage.save_checkpoint(checkpoint)
logger.info(f"Created checkpoint {checkpoint_id} for workflow {self._workflow_id}")
return checkpoint_id
async def load_checkpoint(self, checkpoint_id: str) -> WorkflowCheckpoint | None:
if not self._checkpoint_storage:
storage = self._get_effective_checkpoint_storage()
if not storage:
raise ValueError("Checkpoint storage not configured")
return await self._checkpoint_storage.load_checkpoint(checkpoint_id)
return await storage.load_checkpoint(checkpoint_id)
def reset_for_new_run(self) -> None:
"""Reset the context for a new workflow run.
This clears messages, events, and resets streaming flag.
Runtime checkpoint storage is NOT cleared here as it's managed at the workflow level.
"""
self._messages.clear()
# Clear any pending events (best-effort) by recreating the queue
@@ -12,10 +12,6 @@ from ._typing_utils import is_type_compatible
logger = logging.getLogger(__name__)
# Track cycle signatures we've already reported to avoid spamming logs when workflows
# with intentional feedback loops are constructed multiple times in the same process.
_LOGGED_CYCLE_SIGNATURES: set[tuple[str, ...]] = set()
# region Enums and Base Classes
class ValidationTypeEnum(Enum):
@@ -168,7 +164,6 @@ class WorkflowGraphValidator:
self._validate_graph_connectivity(start_executor_id)
self._validate_self_loops()
self._validate_dead_ends()
self._validate_cycles()
def _validate_handler_output_annotations(self) -> None:
"""Validate that each handler's ctx parameter is annotated with WorkflowContext[T].
@@ -394,96 +389,6 @@ class WorkflowGraphValidator:
f"Verify these are intended as final nodes in the workflow."
)
def _validate_cycles(self) -> None:
"""Detect cycles in the workflow graph.
Cycles might be intentional for iterative processing but should be flagged
for review to ensure proper termination conditions exist. We surface each
distinct cycle group only once per process to avoid noisy, repeated warnings
when rebuilding the same workflow.
"""
# Build adjacency list (ensure every executor appears even if it has no outgoing edges)
graph: dict[str, list[str]] = defaultdict(list)
for edge in self._edges:
graph[edge.source_id].append(edge.target_id)
graph.setdefault(edge.target_id, [])
for executor_id in self._executors:
graph.setdefault(executor_id, [])
# Tarjan's algorithm to locate strongly-connected components that form cycles
index: dict[str, int] = {}
lowlink: dict[str, int] = {}
on_stack: set[str] = set()
stack: list[str] = []
current_index = 0
cycle_components: list[list[str]] = []
def strongconnect(node: str) -> None:
nonlocal current_index
index[node] = current_index
lowlink[node] = current_index
current_index += 1
stack.append(node)
on_stack.add(node)
for neighbor in graph[node]:
if neighbor not in index:
strongconnect(neighbor)
lowlink[node] = min(lowlink[node], lowlink[neighbor])
elif neighbor in on_stack:
lowlink[node] = min(lowlink[node], index[neighbor])
if lowlink[node] == index[node]:
component: list[str] = []
while True:
member = stack.pop()
on_stack.discard(member)
component.append(member)
if member == node:
break
# A strongly connected component represents a cycle if it has more than one
# node or if a single node references itself directly.
if len(component) > 1 or any(member in graph[member] for member in component):
cycle_components.append(component)
for executor_id in graph:
if executor_id not in index:
strongconnect(executor_id)
if not cycle_components:
return
unseen_components: list[list[str]] = []
for component in cycle_components:
signature = tuple(sorted(component))
if signature in _LOGGED_CYCLE_SIGNATURES:
continue
_LOGGED_CYCLE_SIGNATURES.add(signature)
unseen_components.append(component)
if not unseen_components:
# All cycles already reported in this process; keep noise low but retain traceability.
logger.debug(
"Cycle detected in workflow graph but previously reported. Components: %s",
[sorted(component) for component in cycle_components],
)
return
def _format_cycle(component: list[str]) -> str:
if not component:
return ""
ordered = list(component)
ordered.append(component[0])
return " -> ".join(ordered)
formatted_cycles = ", ".join(_format_cycle(component) for component in unseen_components)
logger.warning(
"Cycle detected in the workflow graph involving: %s. Ensure termination or iteration limits exist.",
formatted_cycles,
)
# endregion
@@ -131,10 +131,16 @@ class Workflow(DictConvertible):
Access these via the input_types and output_types properties.
## Execution Methods
- run(): Execute to completion, returns WorkflowRunResult with all events
- run_stream(): Returns async generator yielding events as they occur
- run_from_checkpoint(): Resume from a saved checkpoint
- run_stream_from_checkpoint(): Resume from checkpoint with streaming
The workflow provides two primary execution APIs, each supporting multiple scenarios:
- **run()**: Execute to completion, returns WorkflowRunResult with all events
- **run_stream()**: Returns async generator yielding events as they occur
Both methods support:
- Initial workflow runs: Provide `message` parameter
- Checkpoint restoration: Provide `checkpoint_id` (and optionally `checkpoint_storage`)
- HIL continuation: Provide `responses` to continue after RequestInfoExecutor requests
- Runtime checkpointing: Provide `checkpoint_storage` to enable/override checkpointing for this run
## External Input Requests
Executors within a workflow can request external input using `ctx.request_info()`:
@@ -142,10 +148,18 @@ class Workflow(DictConvertible):
2. Executor implements `response_handler()` to process the response
3. Requests are emitted as RequestInfoEvent instances in the event stream
4. Workflow enters IDLE_WITH_PENDING_REQUESTS state
5. Caller handles requests and uses send_responses()/send_responses_streaming() to continue
6. Responses are routed back to the requesting executors and response handlers are invoked
5. Caller handles requests and provides responses via the `send_responses` or `send_responses_streaming` methods
6. Responses are routed to the requesting executors and response handlers are invoked
## Checkpointing
Checkpointing can be configured at build time or runtime:
Build-time (via WorkflowBuilder):
workflow = WorkflowBuilder().with_checkpointing(storage).build()
Runtime (via run/run_stream parameters):
result = await workflow.run(message, checkpoint_storage=runtime_storage)
When enabled, checkpoints are created at the end of each superstep, capturing:
- Executor states
- Messages in transit
@@ -370,65 +384,146 @@ class Workflow(DictConvertible):
capture_exception(span, exception=exc)
raise
# region Streaming Run
async def run_stream(self, message: Any) -> AsyncIterable[WorkflowEvent]:
"""Run the workflow with a starting message and stream events.
async def _execute_with_message_or_checkpoint(
self,
message: Any | None,
checkpoint_id: str | None,
checkpoint_storage: CheckpointStorage | None,
) -> None:
"""Internal handler for executing workflow with either initial message or checkpoint restoration.
Args:
message: The message to be sent to the starting executor.
message: Initial message for the start executor (for new runs).
checkpoint_id: ID of checkpoint to restore from (for resuming runs).
checkpoint_storage: Runtime checkpoint storage.
Yields:
WorkflowEvent: The events generated during the workflow execution.
Raises:
ValueError: If both message and checkpoint_id are None (nothing to execute).
"""
self._ensure_not_running()
# Validate that we have something to execute
if message is None and checkpoint_id is None:
raise ValueError("Must provide either 'message' or 'checkpoint_id'")
async def initial_execution() -> None:
# Handle checkpoint restoration
if checkpoint_id is not None:
has_checkpointing = self._runner.context.has_checkpointing()
if not has_checkpointing and checkpoint_storage is None:
raise ValueError(
"Cannot restore from checkpoint: either provide checkpoint_storage parameter "
"or build workflow with WorkflowBuilder.with_checkpointing(checkpoint_storage)."
)
restored = await self._runner.restore_from_checkpoint(checkpoint_id, checkpoint_storage)
if not restored:
raise RuntimeError(f"Failed to restore from checkpoint: {checkpoint_id}")
# Handle initial message
elif message is not None:
executor = self.get_start_executor()
await executor.execute(
message,
[self.__class__.__name__], # source_executor_ids
self._shared_state, # shared_state
self._runner.context, # runner_context
trace_contexts=None, # No parent trace context for workflow start
source_span_ids=None, # No source span for workflow start
[self.__class__.__name__],
self._shared_state,
self._runner.context,
trace_contexts=None,
source_span_ids=None,
)
try:
async for event in self._run_workflow_with_tracing(
initial_executor_fn=initial_execution, reset_context=True, streaming=True
):
yield event
finally:
self._reset_running_flag()
async def run_stream_from_checkpoint(
async def run_stream(
self,
checkpoint_id: str,
message: Any | None = None,
*,
checkpoint_id: str | None = None,
checkpoint_storage: CheckpointStorage | None = None,
) -> AsyncIterable[WorkflowEvent]:
"""Resume workflow execution from a checkpoint and stream events.
"""Run the workflow and stream events.
Unified streaming interface supporting initial runs and checkpoint restoration.
Args:
checkpoint_id: The ID of the checkpoint to restore from.
checkpoint_storage: Optional checkpoint storage to use for restoration.
If not provided, the workflow must have been built with checkpointing enabled.
message: Initial message for the start executor. Required for new workflow runs,
should be None when resuming from checkpoint.
checkpoint_id: ID of checkpoint to restore from. If provided, the workflow resumes
from this checkpoint instead of starting fresh. When resuming, checkpoint_storage
must be provided (either at build time or runtime) to load the checkpoint.
checkpoint_storage: Runtime checkpoint storage with two behaviors:
- With checkpoint_id: Used to load and restore the specified checkpoint
- Without checkpoint_id: Enables checkpointing for this run, overriding
build-time configuration
Yields:
WorkflowEvent: Events generated during workflow execution.
Raises:
ValueError: If neither checkpoint_storage is provided nor checkpointing is enabled.
ValueError: If both message and checkpoint_id are provided, or if neither is provided.
ValueError: If checkpoint_id is provided but no checkpoint storage is available
(neither at build time nor runtime).
RuntimeError: If checkpoint restoration fails.
Examples:
Initial run:
.. code-block:: python
async for event in workflow.run_stream("start message"):
process(event)
Enable checkpointing at runtime:
.. code-block:: python
storage = FileCheckpointStorage("./checkpoints")
async for event in workflow.run_stream("start", checkpoint_storage=storage):
process(event)
Resume from checkpoint (storage provided at build time):
.. code-block:: python
async for event in workflow.run_stream(checkpoint_id="cp_123"):
process(event)
Resume from checkpoint (storage provided at runtime):
.. code-block:: python
storage = FileCheckpointStorage("./checkpoints")
async for event in workflow.run_stream(checkpoint_id="cp_123", checkpoint_storage=storage):
process(event)
"""
# Validate mutually exclusive parameters BEFORE setting running flag
if message is not None and checkpoint_id is not None:
raise ValueError("Cannot provide both 'message' and 'checkpoint_id'. Use one or the other.")
if message is None and checkpoint_id is None:
raise ValueError("Must provide either 'message' (new run) or 'checkpoint_id' (resume).")
self._ensure_not_running()
# Enable runtime checkpointing if storage provided
# Two cases:
# 1. checkpoint_storage + checkpoint_id: Load checkpoint from this storage and resume
# 2. checkpoint_storage without checkpoint_id: Enable checkpointing for this run
if checkpoint_storage is not None:
self._runner.context.set_runtime_checkpoint_storage(checkpoint_storage)
try:
# Reset context only for new runs (not checkpoint restoration)
reset_context = message is not None and checkpoint_id is None
async for event in self._run_workflow_with_tracing(
initial_executor_fn=functools.partial(self._checkpoint_restoration, checkpoint_id, checkpoint_storage),
reset_context=False, # Don't reset context when resuming from checkpoint
initial_executor_fn=functools.partial(
self._execute_with_message_or_checkpoint, message, checkpoint_id, checkpoint_storage
),
reset_context=reset_context,
streaming=True,
):
yield event
finally:
if checkpoint_storage is not None:
self._runner.context.clear_runtime_checkpoint_storage()
self._reset_running_flag()
async def send_responses_streaming(self, responses: dict[str, Any]) -> AsyncIterable[WorkflowEvent]:
@@ -452,42 +547,96 @@ class Workflow(DictConvertible):
finally:
self._reset_running_flag()
# endregion: Streaming Run
async def run(
self,
message: Any | None = None,
*,
checkpoint_id: str | None = None,
checkpoint_storage: CheckpointStorage | None = None,
include_status_events: bool = False,
) -> WorkflowRunResult:
"""Run the workflow to completion and return all events.
# region: Run
async def run(self, message: Any, *, include_status_events: bool = False) -> WorkflowRunResult:
"""Run the workflow with the given message.
Unified non-streaming interface supporting initial runs and checkpoint restoration.
Args:
message: The message to be processed by the workflow.
message: Initial message for the start executor. Required for new workflow runs,
should be None when resuming from checkpoint.
checkpoint_id: ID of checkpoint to restore from. If provided, the workflow resumes
from this checkpoint instead of starting fresh. When resuming, checkpoint_storage
must be provided (either at build time or runtime) to load the checkpoint.
checkpoint_storage: Runtime checkpoint storage with two behaviors:
- With checkpoint_id: Used to load and restore the specified checkpoint
- Without checkpoint_id: Enables checkpointing for this run, overriding
build-time configuration
include_status_events: Whether to include WorkflowStatusEvent instances in the result list.
Returns:
A WorkflowRunResult instance containing a list of events generated during the workflow execution.
"""
self._ensure_not_running()
try:
A WorkflowRunResult instance containing events generated during workflow execution.
async def initial_execution() -> None:
executor = self.get_start_executor()
await executor.execute(
message,
[self.__class__.__name__], # source_executor_ids
self._shared_state, # shared_state
self._runner.context, # runner_context
trace_contexts=None, # No parent trace context for workflow start
source_span_ids=None, # No source span for workflow start
)
Raises:
ValueError: If both message and checkpoint_id are provided, or if neither is provided.
ValueError: If checkpoint_id is provided but no checkpoint storage is available
(neither at build time nor runtime).
RuntimeError: If checkpoint restoration fails.
Examples:
Initial run:
.. code-block:: python
result = await workflow.run("start message")
outputs = result.get_outputs()
Enable checkpointing at runtime:
.. code-block:: python
storage = FileCheckpointStorage("./checkpoints")
result = await workflow.run("start", checkpoint_storage=storage)
Resume from checkpoint (storage provided at build time):
.. code-block:: python
result = await workflow.run(checkpoint_id="cp_123")
Resume from checkpoint (storage provided at runtime):
.. code-block:: python
storage = FileCheckpointStorage("./checkpoints")
result = await workflow.run(checkpoint_id="cp_123", checkpoint_storage=storage)
"""
# Validate mutually exclusive parameters BEFORE setting running flag
if message is not None and checkpoint_id is not None:
raise ValueError("Cannot provide both 'message' and 'checkpoint_id'. Use one or the other.")
if message is None and checkpoint_id is None:
raise ValueError("Must provide either 'message' (new run) or 'checkpoint_id' (resume).")
self._ensure_not_running()
# Enable runtime checkpointing if storage provided
if checkpoint_storage is not None:
self._runner.context.set_runtime_checkpoint_storage(checkpoint_storage)
try:
# Reset context only for new runs (not checkpoint restoration)
reset_context = message is not None and checkpoint_id is None
raw_events = [
event
async for event in self._run_workflow_with_tracing(
initial_executor_fn=initial_execution,
reset_context=True,
initial_executor_fn=functools.partial(
self._execute_with_message_or_checkpoint, message, checkpoint_id, checkpoint_storage
),
reset_context=reset_context,
)
]
finally:
if checkpoint_storage is not None:
self._runner.context.clear_runtime_checkpoint_storage()
self._reset_running_flag()
# Filter events for non-streaming mode
@@ -508,42 +657,6 @@ class Workflow(DictConvertible):
return WorkflowRunResult(filtered, status_events)
async def run_from_checkpoint(
self,
checkpoint_id: str,
checkpoint_storage: CheckpointStorage | None = None,
) -> WorkflowRunResult:
"""Resume workflow execution from a checkpoint.
Args:
checkpoint_id: The ID of the checkpoint to restore from.
checkpoint_storage: Optional checkpoint storage to use for restoration.
If not provided, the workflow must have been built with checkpointing enabled.
Returns:
A WorkflowRunResult instance containing a list of events generated during the workflow execution.
Raises:
ValueError: If neither checkpoint_storage is provided nor checkpointing is enabled.
RuntimeError: If checkpoint restoration fails.
"""
self._ensure_not_running()
try:
events = [
event
async for event in self._run_workflow_with_tracing(
initial_executor_fn=functools.partial(
self._checkpoint_restoration, checkpoint_id, checkpoint_storage
),
reset_context=False, # Don't reset context when resuming from checkpoint
)
]
status_events = [e for e in events if isinstance(e, WorkflowStatusEvent)]
filtered_events = [e for e in events if not isinstance(e, (WorkflowStatusEvent, WorkflowStartedEvent))]
return WorkflowRunResult(filtered_events, status_events)
finally:
self._reset_running_flag()
async def send_responses(self, responses: dict[str, Any]) -> WorkflowRunResult:
"""Send responses back to the workflow.
@@ -568,8 +681,6 @@ class Workflow(DictConvertible):
finally:
self._reset_running_flag()
# endregion: Run
async def _send_responses_internal(self, responses: dict[str, Any]) -> None:
"""Internal method to validate and send responses to the executors."""
pending_requests = await self._runner_context.get_pending_request_info_events()
@@ -592,21 +703,6 @@ class Workflow(DictConvertible):
for request_id, response in responses.items()
])
async def _checkpoint_restoration(self, checkpoint_id: str, checkpoint_storage: CheckpointStorage | None) -> None:
"""Internal method to restore a run from a checkpoint."""
has_checkpointing = self._runner.context.has_checkpointing()
if not has_checkpointing and checkpoint_storage is None:
raise ValueError(
"Cannot restore from checkpoint: either provide checkpoint_storage parameter "
"or build workflow with WorkflowBuilder.with_checkpointing(checkpoint_storage)."
)
restored = await self._runner.restore_from_checkpoint(checkpoint_id, checkpoint_storage)
if not restored:
raise RuntimeError(f"Failed to restore from checkpoint: {checkpoint_id}")
def _get_executor_by_id(self, executor_id: str) -> Executor:
"""Get an executor by its ID.
@@ -525,7 +525,8 @@ class WorkflowExecutor(Executor):
for request_info_event in execution_context.pending_requests.values()
]
await asyncio.gather(*[
self.workflow._runner_context.add_request_info_event(event) for event in request_info_events
self.workflow._runner_context.add_request_info_event(event) # pyright: ignore[reportPrivateUsage]
for event in request_info_events
])
self._state_loaded = True
@@ -0,0 +1,23 @@
# Copyright (c) Microsoft. All rights reserved.
import importlib
from typing import Any
PACKAGE_NAME = "agent_framework_anthropic"
PACKAGE_EXTRA = "anthropic"
_IMPORTS = ["__version__", "AnthropicClient"]
def __getattr__(name: str) -> Any:
if name in _IMPORTS:
try:
return getattr(importlib.import_module(PACKAGE_NAME), name)
except ModuleNotFoundError as exc:
raise ModuleNotFoundError(
f"The '{PACKAGE_EXTRA}' extra is not installed, please do `pip install agent-framework-{PACKAGE_EXTRA}`"
) from exc
raise AttributeError(f"Module {PACKAGE_NAME} has no attribute {name}.")
def __dir__() -> list[str]:
return _IMPORTS
@@ -0,0 +1,5 @@
# Copyright (c) Microsoft. All rights reserved.
from agent_framework_anthropic import AnthropicClient, __version__
__all__ = ["AnthropicClient", "__version__"]
@@ -1413,7 +1413,7 @@ def _capture_messages(
finish_reason: "FinishReason | None" = None,
) -> None:
"""Log messages with extra information."""
from ._clients import prepare_messages
from ._types import prepare_messages
prepped = prepare_messages(messages)
otel_messages: list[dict[str, Any]] = []
@@ -408,7 +408,7 @@ class OpenAIAssistantsClient(OpenAIConfigMixin, BaseChatClient):
run_options["tools"] = tool_definitions
if chat_options.tool_choice == "none" or chat_options.tool_choice == "auto":
run_options["tool_choice"] = chat_options.tool_choice
run_options["tool_choice"] = chat_options.tool_choice.mode
elif (
isinstance(chat_options.tool_choice, ToolMode)
and chat_options.tool_choice == "required"
@@ -191,6 +191,8 @@ class OpenAIBaseChatClient(OpenAIBase, BaseChatClient):
for key, value in additional_properties.items():
if value is not None:
options_dict[key] = value
if (tool_choice := options_dict.get("tool_choice")) and len(tool_choice.keys()) == 1:
options_dict["tool_choice"] = tool_choice["mode"]
return options_dict
def _create_chat_response(self, response: ChatCompletion, chat_options: ChatOptions) -> "ChatResponse":
@@ -345,6 +345,8 @@ class OpenAIBaseResponsesClient(OpenAIBase, BaseChatClient):
options_dict[key] = value
if "store" not in options_dict:
options_dict["store"] = False
if (tool_choice := options_dict.get("tool_choice")) and len(tool_choice.keys()) == 1:
options_dict["tool_choice"] = tool_choice["mode"]
return options_dict
def _prepare_chat_messages_for_request(self, chat_messages: Sequence[ChatMessage]) -> list[dict[str, Any]]:
+3 -1
View File
@@ -4,7 +4,7 @@ description = "Microsoft Agent Framework for building AI Agents with Python. Thi
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
readme = "README.md"
requires-python = ">=3.10"
version = "1.0.0b251028"
version = "1.0.0b251104"
license-files = ["LICENSE"]
urls.homepage = "https://aka.ms/agent-framework"
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -45,6 +45,8 @@ all = [
"agent-framework-mem0",
"agent-framework-redis",
"agent-framework-devui",
"agent-framework-purview",
"agent-framework-anthropic",
]
[tool.uv]
@@ -15,7 +15,6 @@ from agent_framework import (
ChatAgent,
ChatClientProtocol,
ChatMessage,
ChatOptions,
ChatResponse,
ChatResponseUpdate,
HostedCodeInterpreterTool,
@@ -155,18 +154,6 @@ def test_azure_assistants_client_init_with_default_headers(azure_openai_unit_tes
assert chat_client.client.default_headers[key] == value
def test_azure_assistants_client_instructions_sent_once(mock_async_azure_openai: MagicMock) -> None:
"""Ensure instructions are only included once for Azure OpenAI Assistants requests."""
chat_client = create_test_azure_assistants_client(mock_async_azure_openai)
instructions = "You are a helpful assistant."
chat_options = ChatOptions(instructions=instructions)
prepared_messages = chat_client.prepare_messages([ChatMessage(role="user", text="Hello")], chat_options)
run_options, _ = chat_client._prepare_options(prepared_messages, chat_options) # type: ignore[reportPrivateUsage]
assert run_options.get("instructions") == instructions
async def test_azure_assistants_client_get_assistant_id_or_create_existing_assistant(
mock_async_azure_openai: MagicMock,
) -> None:
@@ -23,7 +23,6 @@ from agent_framework import (
ChatAgent,
ChatClientProtocol,
ChatMessage,
ChatOptions,
ChatResponse,
ChatResponseUpdate,
TextContent,
@@ -84,18 +83,6 @@ def test_init_base_url(azure_openai_unit_test_env: dict[str, str]) -> None:
assert azure_chat_client.client.default_headers[key] == value
def test_azure_openai_chat_client_instructions_sent_once(azure_openai_unit_test_env: dict[str, str]) -> None:
"""Ensure instructions are only included once when preparing Azure OpenAI chat requests."""
client = AzureOpenAIChatClient()
instructions = "You are a helpful assistant."
chat_options = ChatOptions(instructions=instructions)
prepared_messages = client.prepare_messages([ChatMessage(role="user", text="Hello")], chat_options)
request_options = client._prepare_options(prepared_messages, chat_options) # type: ignore[reportPrivateUsage]
assert json.dumps(request_options).count(instructions) == 1
@pytest.mark.parametrize("exclude_list", [["AZURE_OPENAI_BASE_URL"]], indirect=True)
def test_init_endpoint(azure_openai_unit_test_env: dict[str, str]) -> None:
azure_chat_client = AzureOpenAIChatClient()
@@ -1,6 +1,5 @@
# Copyright (c) Microsoft. All rights reserved.
import json
import os
from typing import Annotated
@@ -15,7 +14,6 @@ from agent_framework import (
ChatAgent,
ChatClientProtocol,
ChatMessage,
ChatOptions,
ChatResponse,
ChatResponseUpdate,
HostedCodeInterpreterTool,
@@ -114,18 +112,6 @@ def test_init_with_default_header(azure_openai_unit_test_env: dict[str, str]) ->
assert azure_responses_client.client.default_headers[key] == value
def test_azure_responses_client_instructions_sent_once(azure_openai_unit_test_env: dict[str, str]) -> None:
"""Ensure instructions are only included once for Azure OpenAI Responses requests."""
client = AzureOpenAIResponsesClient()
instructions = "You are a helpful assistant."
chat_options = ChatOptions(instructions=instructions)
prepared_messages = client.prepare_messages([ChatMessage(role="user", text="Hello")], chat_options)
request_options = client._prepare_options(prepared_messages, chat_options) # type: ignore[reportPrivateUsage]
assert json.dumps(request_options).count(instructions) == 1
@pytest.mark.parametrize("exclude_list", [["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"]], indirect=True)
def test_init_with_empty_model_id(azure_openai_unit_test_env: dict[str, str]) -> None:
with pytest.raises(ServiceInitializationError):
@@ -1,10 +1,13 @@
# Copyright (c) Microsoft. All rights reserved.
from unittest.mock import patch
from agent_framework import (
BaseChatClient,
ChatClientProtocol,
ChatMessage,
ChatOptions,
Role,
)
@@ -39,3 +42,20 @@ async def test_base_client_get_response(chat_client_base: ChatClientProtocol):
async def test_base_client_get_streaming_response(chat_client_base: ChatClientProtocol):
async for update in chat_client_base.get_streaming_response(ChatMessage(role="user", text="Hello")):
assert update.text == "update - Hello" or update.text == "another update"
async def test_chat_client_instructions_handling(chat_client_base: ChatClientProtocol):
instructions = "You are a helpful assistant."
with patch.object(
chat_client_base,
"_inner_get_response",
) as mock_inner_get_response:
await chat_client_base.get_response("hello", chat_options=ChatOptions(instructions=instructions))
mock_inner_get_response.assert_called_once()
_, kwargs = mock_inner_get_response.call_args
messages = kwargs.get("messages", [])
assert len(messages) == 2
assert messages[0].role == Role.SYSTEM
assert messages[0].text == instructions
assert messages[1].role == Role.USER
assert messages[1].text == "hello"
@@ -193,18 +193,6 @@ def test_openai_assistants_client_init_with_default_headers(openai_unit_test_env
assert chat_client.client.default_headers[key] == value
def test_openai_assistants_client_instructions_sent_once(mock_async_openai: MagicMock) -> None:
"""Ensure instructions are only included once for OpenAI Assistants requests."""
chat_client = create_test_openai_assistants_client(mock_async_openai)
instructions = "You are a helpful assistant."
chat_options = ChatOptions(instructions=instructions)
prepared_messages = chat_client.prepare_messages([ChatMessage(role=Role.USER, text="Hello")], chat_options)
run_options, _ = chat_client._prepare_options(prepared_messages, chat_options) # type: ignore[reportPrivateUsage]
assert run_options.get("instructions") == instructions
async def test_openai_assistants_client_get_assistant_id_or_create_existing_assistant(
mock_async_openai: MagicMock,
) -> None:
@@ -1,6 +1,5 @@
# Copyright (c) Microsoft. All rights reserved.
import json
import os
from typing import Annotated
from unittest.mock import MagicMock, patch
@@ -100,18 +99,6 @@ def test_init_base_url_from_settings_env() -> None:
assert str(client.client.base_url) == "https://custom-openai-endpoint.com/v1/"
def test_openai_chat_client_instructions_sent_once(openai_unit_test_env: dict[str, str]) -> None:
"""Ensure instructions are only included once for OpenAI chat requests."""
client = OpenAIChatClient()
instructions = "You are a helpful assistant."
chat_options = ChatOptions(instructions=instructions)
prepared_messages = client.prepare_messages([ChatMessage(role="user", text="Hello")], chat_options)
request_options = client._prepare_options(prepared_messages, chat_options) # type: ignore[reportPrivateUsage]
assert json.dumps(request_options).count(instructions) == 1
@pytest.mark.parametrize("exclude_list", [["OPENAI_CHAT_MODEL_ID"]], indirect=True)
def test_init_with_empty_model_id(openai_unit_test_env: dict[str, str]) -> None:
with pytest.raises(ServiceInitializationError):
@@ -2,7 +2,6 @@
import asyncio
import base64
import json
import os
from typing import Annotated
from unittest.mock import MagicMock, patch
@@ -138,18 +137,6 @@ def test_init_with_default_header(openai_unit_test_env: dict[str, str]) -> None:
assert openai_responses_client.client.default_headers[key] == value
def test_openai_responses_client_instructions_sent_once(openai_unit_test_env: dict[str, str]) -> None:
"""Ensure instructions are only included once for OpenAI Responses requests."""
client = OpenAIResponsesClient()
instructions = "You are a helpful assistant."
chat_options = ChatOptions(instructions=instructions)
prepared_messages = client.prepare_messages([ChatMessage(role="user", text="Hello")], chat_options)
request_options = client._prepare_options(prepared_messages, chat_options) # type: ignore[reportPrivateUsage]
assert json.dumps(request_options).count(instructions) == 1
@pytest.mark.parametrize("exclude_list", [["OPENAI_RESPONSES_MODEL_ID"]], indirect=True)
def test_init_with_empty_model_id(openai_unit_test_env: dict[str, str]) -> None:
with pytest.raises(ServiceInitializationError):
@@ -125,7 +125,7 @@ async def test_agent_executor_checkpoint_stores_and_restores_state() -> None:
# Resume from checkpoint
resumed_output: AgentExecutorResponse | None = None
async for ev in wf_resume.run_stream_from_checkpoint(restore_checkpoint.checkpoint_id):
async for ev in wf_resume.run_stream(checkpoint_id=restore_checkpoint.checkpoint_id):
if isinstance(ev, WorkflowOutputEvent):
resumed_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state in (
@@ -46,8 +46,8 @@ async def test_resume_fails_when_graph_mismatch() -> None:
with pytest.raises(ValueError, match="Workflow graph has changed"):
_ = [
event
async for event in mismatched_workflow.run_stream_from_checkpoint(
target_checkpoint.checkpoint_id,
async for event in mismatched_workflow.run_stream(
checkpoint_id=target_checkpoint.checkpoint_id,
checkpoint_storage=storage,
)
]
@@ -65,8 +65,8 @@ async def test_resume_succeeds_when_graph_matches() -> None:
events = [
event
async for event in resumed_workflow.run_stream_from_checkpoint(
target_checkpoint.checkpoint_id,
async for event in resumed_workflow.run_stream(
checkpoint_id=target_checkpoint.checkpoint_id,
checkpoint_storage=storage,
)
]
@@ -195,7 +195,7 @@ async def test_concurrent_checkpoint_resume_round_trip() -> None:
wf_resume = ConcurrentBuilder().participants(list(resumed_participants)).with_checkpointing(storage).build()
resumed_output: list[ChatMessage] | None = None
async for ev in wf_resume.run_stream_from_checkpoint(resume_checkpoint.checkpoint_id):
async for ev in wf_resume.run_stream(checkpoint_id=resume_checkpoint.checkpoint_id):
if isinstance(ev, WorkflowOutputEvent):
resumed_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state in (
@@ -207,3 +207,74 @@ async def test_concurrent_checkpoint_resume_round_trip() -> None:
assert resumed_output is not None
assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
assert [m.text for m in resumed_output] == [m.text for m in baseline_output]
async def test_concurrent_checkpoint_runtime_only() -> None:
"""Test checkpointing configured ONLY at runtime, not at build time."""
storage = InMemoryCheckpointStorage()
agents = [_FakeAgentExec(id="agent1", reply_text="A1"), _FakeAgentExec(id="agent2", reply_text="A2")]
wf = ConcurrentBuilder().participants(agents).build()
baseline_output: list[ChatMessage] | None = None
async for ev in wf.run_stream("runtime checkpoint test", checkpoint_storage=storage):
if isinstance(ev, WorkflowOutputEvent):
baseline_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
break
assert baseline_output is not None
checkpoints = await storage.list_checkpoints()
assert checkpoints
checkpoints.sort(key=lambda cp: cp.timestamp)
resume_checkpoint = next(
(cp for cp in checkpoints if (cp.metadata or {}).get("checkpoint_type") == "superstep"),
checkpoints[-1],
)
resumed_agents = [_FakeAgentExec(id="agent1", reply_text="A1"), _FakeAgentExec(id="agent2", reply_text="A2")]
wf_resume = ConcurrentBuilder().participants(resumed_agents).build()
resumed_output: list[ChatMessage] | None = None
async for ev in wf_resume.run_stream(checkpoint_id=resume_checkpoint.checkpoint_id, checkpoint_storage=storage):
if isinstance(ev, WorkflowOutputEvent):
resumed_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
assert resumed_output is not None
assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
async def test_concurrent_checkpoint_runtime_overrides_buildtime() -> None:
"""Test that runtime checkpoint storage overrides build-time configuration."""
import tempfile
with tempfile.TemporaryDirectory() as temp_dir1, tempfile.TemporaryDirectory() as temp_dir2:
from agent_framework._workflows._checkpoint import FileCheckpointStorage
buildtime_storage = FileCheckpointStorage(temp_dir1)
runtime_storage = FileCheckpointStorage(temp_dir2)
agents = [_FakeAgentExec(id="agent1", reply_text="A1"), _FakeAgentExec(id="agent2", reply_text="A2")]
wf = ConcurrentBuilder().participants(agents).with_checkpointing(buildtime_storage).build()
baseline_output: list[ChatMessage] | None = None
async for ev in wf.run_stream("override test", checkpoint_storage=runtime_storage):
if isinstance(ev, WorkflowOutputEvent):
baseline_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
break
assert baseline_output is not None
buildtime_checkpoints = await buildtime_storage.list_checkpoints()
runtime_checkpoints = await runtime_storage.list_checkpoints()
assert len(runtime_checkpoints) > 0, "Runtime storage should have checkpoints"
assert len(buildtime_checkpoints) == 0, "Build-time storage should have no checkpoints when overridden"
@@ -742,3 +742,73 @@ class TestRoundLimitEnforcement:
# The last message should be about round limit
final_output = outputs[-1]
assert "round limit" in final_output.text.lower()
async def test_group_chat_checkpoint_runtime_only() -> None:
"""Test checkpointing configured ONLY at runtime, not at build time."""
from agent_framework import WorkflowRunState, WorkflowStatusEvent
storage = InMemoryCheckpointStorage()
agent_a = StubAgent("agentA", "Reply from A")
agent_b = StubAgent("agentB", "Reply from B")
selector = make_sequence_selector()
wf = GroupChatBuilder().participants([agent_a, agent_b]).select_speakers(selector).build()
baseline_output: list[ChatMessage] | None = None
async for ev in wf.run_stream("runtime checkpoint test", checkpoint_storage=storage):
if isinstance(ev, WorkflowOutputEvent):
baseline_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
assert baseline_output is not None
checkpoints = await storage.list_checkpoints()
assert len(checkpoints) > 0, "Runtime-only checkpointing should have created checkpoints"
async def test_group_chat_checkpoint_runtime_overrides_buildtime() -> None:
"""Test that runtime checkpoint storage overrides build-time configuration."""
import tempfile
with tempfile.TemporaryDirectory() as temp_dir1, tempfile.TemporaryDirectory() as temp_dir2:
from agent_framework import WorkflowRunState, WorkflowStatusEvent
from agent_framework._workflows._checkpoint import FileCheckpointStorage
buildtime_storage = FileCheckpointStorage(temp_dir1)
runtime_storage = FileCheckpointStorage(temp_dir2)
agent_a = StubAgent("agentA", "Reply from A")
agent_b = StubAgent("agentB", "Reply from B")
selector = make_sequence_selector()
wf = (
GroupChatBuilder()
.participants([agent_a, agent_b])
.select_speakers(selector)
.with_checkpointing(buildtime_storage)
.build()
)
baseline_output: list[ChatMessage] | None = None
async for ev in wf.run_stream("override test", checkpoint_storage=runtime_storage):
if isinstance(ev, WorkflowOutputEvent):
baseline_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
assert baseline_output is not None
buildtime_checkpoints = await buildtime_storage.list_checkpoints()
runtime_checkpoints = await runtime_storage.list_checkpoints()
assert len(runtime_checkpoints) > 0, "Runtime storage should have checkpoints"
assert len(buildtime_checkpoints) == 0, "Build-time storage should have no checkpoints when overridden"
@@ -3,6 +3,7 @@
from collections.abc import AsyncIterable, AsyncIterator
from dataclasses import dataclass
from typing import Any, cast
from unittest.mock import MagicMock
import pytest
@@ -10,6 +11,7 @@ from agent_framework import (
AgentRunResponse,
AgentRunResponseUpdate,
BaseAgent,
ChatAgent,
ChatMessage,
FunctionCallContent,
HandoffBuilder,
@@ -20,6 +22,8 @@ from agent_framework import (
WorkflowEvent,
WorkflowOutputEvent,
)
from agent_framework._mcp import MCPTool
from agent_framework._workflows._handoff import _clone_chat_agent
@dataclass
@@ -155,31 +159,6 @@ async def _drain(stream: AsyncIterable[WorkflowEvent]) -> list[WorkflowEvent]:
return [event async for event in stream]
async def test_handoff_routes_to_specialist_and_requests_user_input():
triage = _RecordingAgent(name="triage", handoff_to="specialist")
specialist = _RecordingAgent(name="specialist")
workflow = HandoffBuilder(participants=[triage, specialist]).set_coordinator("triage").build()
events = await _drain(workflow.run_stream("Need help with a refund"))
assert triage.calls, "Starting agent should receive initial conversation"
assert specialist.calls, "Specialist should be invoked after handoff"
assert len(specialist.calls[0]) == 2 # user + triage reply
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert requests, "Workflow should request additional user input"
request_payload = requests[-1].data
assert isinstance(request_payload, HandoffUserInputRequest)
assert len(request_payload.conversation) == 4 # user, triage tool call, tool ack, specialist
assert request_payload.conversation[2].role == Role.TOOL
assert request_payload.conversation[3].role == Role.ASSISTANT
assert "specialist reply" in request_payload.conversation[3].text
follow_up = await _drain(workflow.send_responses_streaming({requests[-1].request_id: "Thanks"}))
assert any(isinstance(ev, RequestInfoEvent) for ev in follow_up)
async def test_specialist_to_specialist_handoff():
"""Test that specialists can hand off to other specialists via .add_handoff() configuration."""
triage = _RecordingAgent(name="triage", handoff_to="specialist")
@@ -393,3 +372,32 @@ async def test_handoff_async_termination_condition() -> None:
user_messages = [msg for msg in final_conv_list if msg.role == Role.USER]
assert len(user_messages) == 2
assert termination_call_count > 0
async def test_clone_chat_agent_preserves_mcp_tools() -> None:
"""Test that _clone_chat_agent preserves MCP tools when cloning an agent."""
mock_chat_client = MagicMock()
mock_mcp_tool = MagicMock(spec=MCPTool)
mock_mcp_tool.name = "test_mcp_tool"
def sample_function() -> str:
return "test"
original_agent = ChatAgent(
chat_client=mock_chat_client,
name="TestAgent",
instructions="Test instructions",
tools=[mock_mcp_tool, sample_function],
)
assert hasattr(original_agent, "_local_mcp_tools")
assert len(original_agent._local_mcp_tools) == 1
assert original_agent._local_mcp_tools[0] == mock_mcp_tool
cloned_agent = _clone_chat_agent(original_agent)
assert hasattr(cloned_agent, "_local_mcp_tools")
assert len(cloned_agent._local_mcp_tools) == 1
assert cloned_agent._local_mcp_tools[0] == mock_mcp_tool
assert len(cloned_agent.chat_options.tools) == 1
@@ -185,6 +185,7 @@ async def test_standard_manager_progress_ledger_and_fallback():
assert ledger2.is_request_satisfied.answer is False
@pytest.mark.skip(reason="Response handling refactored - responses no longer passed to run_stream()")
async def test_magentic_workflow_plan_review_approval_to_completion():
manager = FakeManager(max_round_count=10)
wf = (
@@ -203,9 +204,9 @@ async def test_magentic_workflow_plan_review_approval_to_completion():
completed = False
output: ChatMessage | None = None
async for ev in wf.send_responses_streaming({
req_event.request_id: MagenticPlanReviewReply(decision=MagenticPlanReviewDecision.APPROVE)
}):
async for ev in wf.run_stream(
responses={req_event.request_id: MagenticPlanReviewReply(decision=MagenticPlanReviewDecision.APPROVE)}
):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
@@ -217,6 +218,7 @@ async def test_magentic_workflow_plan_review_approval_to_completion():
assert isinstance(output, ChatMessage)
@pytest.mark.skip(reason="Response handling refactored - responses no longer passed to run_stream()")
async def test_magentic_plan_review_approve_with_comments_replans_and_proceeds():
class CountingManager(FakeManager):
# Declare as a model field so assignment is allowed under Pydantic
@@ -248,12 +250,14 @@ async def test_magentic_plan_review_approve_with_comments_replans_and_proceeds()
# Reply APPROVE with comments (no edited text). Expect one replan and no second review round.
saw_second_review = False
completed = False
async for ev in wf.send_responses_streaming({
req_event.request_id: MagenticPlanReviewReply(
decision=MagenticPlanReviewDecision.APPROVE,
comments="Looks good; consider Z",
)
}):
async for ev in wf.run_stream(
responses={
req_event.request_id: MagenticPlanReviewReply(
decision=MagenticPlanReviewDecision.APPROVE,
comments="Looks good; consider Z",
)
}
):
if isinstance(ev, RequestInfoEvent) and ev.request_type is MagenticPlanReviewRequest:
saw_second_review = True
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
@@ -294,6 +298,7 @@ async def test_magentic_orchestrator_round_limit_produces_partial_result():
assert data.role == Role.ASSISTANT
@pytest.mark.skip(reason="Response handling refactored - send_responses_streaming no longer exists")
async def test_magentic_checkpoint_resume_round_trip():
storage = InMemoryCheckpointStorage()
@@ -334,7 +339,7 @@ async def test_magentic_checkpoint_resume_round_trip():
reply = MagenticPlanReviewReply(decision=MagenticPlanReviewDecision.APPROVE)
completed: WorkflowOutputEvent | None = None
req_event = None
async for event in wf_resume.run_stream_from_checkpoint(
async for event in wf_resume.run_stream(
resume_checkpoint.checkpoint_id,
):
if isinstance(event, RequestInfoEvent) and event.request_type is MagenticPlanReviewRequest:
@@ -604,7 +609,7 @@ async def test_magentic_checkpoint_resume_inner_loop_superstep():
)
completed: WorkflowOutputEvent | None = None
async for event in resumed.run_stream_from_checkpoint(inner_loop_checkpoint.checkpoint_id): # type: ignore[reportUnknownMemberType]
async for event in resumed.run_stream(checkpoint_id=inner_loop_checkpoint.checkpoint_id): # type: ignore[reportUnknownMemberType]
if isinstance(event, WorkflowOutputEvent):
completed = event
@@ -646,7 +651,7 @@ async def test_magentic_checkpoint_resume_after_reset():
)
completed: WorkflowOutputEvent | None = None
async for event in resumed_workflow.run_stream_from_checkpoint(resumed_state.checkpoint_id):
async for event in resumed_workflow.run_stream(checkpoint_id=resumed_state.checkpoint_id):
if isinstance(event, WorkflowOutputEvent):
completed = event
@@ -687,8 +692,8 @@ async def test_magentic_checkpoint_resume_rejects_participant_renames():
)
with pytest.raises(ValueError, match="Workflow graph has changed"):
async for _ in renamed_workflow.run_stream_from_checkpoint(
target_checkpoint.checkpoint_id, # type: ignore[reportUnknownMemberType]
async for _ in renamed_workflow.run_stream(
checkpoint_id=target_checkpoint.checkpoint_id, # type: ignore[reportUnknownMemberType]
):
pass
@@ -735,3 +740,66 @@ async def test_magentic_stall_and_reset_successfully():
assert isinstance(output_event.data, ChatMessage)
assert output_event.data.text is not None
assert output_event.data.text == "re-ledger"
async def test_magentic_checkpoint_runtime_only() -> None:
"""Test checkpointing configured ONLY at runtime, not at build time."""
storage = InMemoryCheckpointStorage()
manager = FakeManager(max_round_count=10)
manager.satisfied_after_signoff = True
wf = MagenticBuilder().participants(agentA=_DummyExec("agentA")).with_standard_manager(manager).build()
baseline_output: ChatMessage | None = None
async for ev in wf.run_stream("runtime checkpoint test", checkpoint_storage=storage):
if isinstance(ev, WorkflowOutputEvent):
baseline_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
assert baseline_output is not None
checkpoints = await storage.list_checkpoints()
assert len(checkpoints) > 0, "Runtime-only checkpointing should have created checkpoints"
async def test_magentic_checkpoint_runtime_overrides_buildtime() -> None:
"""Test that runtime checkpoint storage overrides build-time configuration."""
import tempfile
with tempfile.TemporaryDirectory() as temp_dir1, tempfile.TemporaryDirectory() as temp_dir2:
from agent_framework._workflows._checkpoint import FileCheckpointStorage
buildtime_storage = FileCheckpointStorage(temp_dir1)
runtime_storage = FileCheckpointStorage(temp_dir2)
manager = FakeManager(max_round_count=10)
manager.satisfied_after_signoff = True
wf = (
MagenticBuilder()
.participants(agentA=_DummyExec("agentA"))
.with_standard_manager(manager)
.with_checkpointing(buildtime_storage)
.build()
)
baseline_output: ChatMessage | None = None
async for ev in wf.run_stream("override test", checkpoint_storage=runtime_storage):
if isinstance(ev, WorkflowOutputEvent):
baseline_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
assert baseline_output is not None
buildtime_checkpoints = await buildtime_storage.list_checkpoints()
runtime_checkpoints = await runtime_storage.list_checkpoints()
assert len(runtime_checkpoints) > 0, "Runtime storage should have checkpoints"
assert len(buildtime_checkpoints) == 0, "Build-time storage should have no checkpoints when overridden"
@@ -378,7 +378,7 @@ class TestRequestInfoAndResponse:
# Step 5: Resume from checkpoint and verify the request can be continued
completed = False
restored_request_event: RequestInfoEvent | None = None
async for event in restored_workflow.run_stream_from_checkpoint(checkpoint_with_request.checkpoint_id):
async for event in restored_workflow.run_stream(checkpoint_id=checkpoint_with_request.checkpoint_id):
# Should re-emit the pending request info event
if isinstance(event, RequestInfoEvent) and event.request_id == request_info_event.request_id:
restored_request_event = event
@@ -145,7 +145,7 @@ async def test_sequential_checkpoint_resume_round_trip() -> None:
wf_resume = SequentialBuilder().participants(list(resumed_agents)).with_checkpointing(storage).build()
resumed_output: list[ChatMessage] | None = None
async for ev in wf_resume.run_stream_from_checkpoint(resume_checkpoint.checkpoint_id):
async for ev in wf_resume.run_stream(checkpoint_id=resume_checkpoint.checkpoint_id):
if isinstance(ev, WorkflowOutputEvent):
resumed_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state in (
@@ -157,3 +157,75 @@ async def test_sequential_checkpoint_resume_round_trip() -> None:
assert resumed_output is not None
assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
assert [m.text for m in resumed_output] == [m.text for m in baseline_output]
async def test_sequential_checkpoint_runtime_only() -> None:
"""Test checkpointing configured ONLY at runtime, not at build time."""
storage = InMemoryCheckpointStorage()
agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
wf = SequentialBuilder().participants(list(agents)).build()
baseline_output: list[ChatMessage] | None = None
async for ev in wf.run_stream("runtime checkpoint test", checkpoint_storage=storage):
if isinstance(ev, WorkflowOutputEvent):
baseline_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
break
assert baseline_output is not None
checkpoints = await storage.list_checkpoints()
assert checkpoints
checkpoints.sort(key=lambda cp: cp.timestamp)
resume_checkpoint = next(
(cp for cp in checkpoints if (cp.metadata or {}).get("checkpoint_type") == "superstep"),
checkpoints[-1],
)
resumed_agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
wf_resume = SequentialBuilder().participants(list(resumed_agents)).build()
resumed_output: list[ChatMessage] | None = None
async for ev in wf_resume.run_stream(checkpoint_id=resume_checkpoint.checkpoint_id, checkpoint_storage=storage):
if isinstance(ev, WorkflowOutputEvent):
resumed_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
assert resumed_output is not None
assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
assert [m.text for m in resumed_output] == [m.text for m in baseline_output]
async def test_sequential_checkpoint_runtime_overrides_buildtime() -> None:
"""Test that runtime checkpoint storage overrides build-time configuration."""
import tempfile
with tempfile.TemporaryDirectory() as temp_dir1, tempfile.TemporaryDirectory() as temp_dir2:
from agent_framework._workflows._checkpoint import FileCheckpointStorage
buildtime_storage = FileCheckpointStorage(temp_dir1)
runtime_storage = FileCheckpointStorage(temp_dir2)
agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
wf = SequentialBuilder().participants(list(agents)).with_checkpointing(buildtime_storage).build()
baseline_output: list[ChatMessage] | None = None
async for ev in wf.run_stream("override test", checkpoint_storage=runtime_storage):
if isinstance(ev, WorkflowOutputEvent):
baseline_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
break
assert baseline_output is not None
buildtime_checkpoints = await buildtime_storage.list_checkpoints()
runtime_checkpoints = await runtime_storage.list_checkpoints()
assert len(runtime_checkpoints) > 0, "Runtime storage should have checkpoints"
assert len(buildtime_checkpoints) == 0, "Build-time storage should have no checkpoints when overridden"
@@ -385,28 +385,6 @@ def test_dead_end_detection(caplog: Any) -> None:
assert "Verify these are intended as final nodes" in caplog.text
def test_cycle_detection_warning(caplog: Any) -> None:
caplog.set_level(logging.WARNING)
executor1 = StringExecutor(id="executor1")
executor2 = StringExecutor(id="executor2")
executor3 = StringExecutor(id="executor3")
# Create a cycle: executor1 -> executor2 -> executor3 -> executor1
workflow = (
WorkflowBuilder()
.add_edge(executor1, executor2)
.add_edge(executor2, executor3)
.add_edge(executor3, executor1)
.set_start_executor(executor1)
.build()
)
assert workflow is not None
assert "Cycle detected in the workflow graph" in caplog.text
assert "Ensure termination or iteration limits exist" in caplog.text
def test_successful_type_compatibility_logging(caplog: Any) -> None:
caplog.set_level(logging.DEBUG)
@@ -420,51 +398,6 @@ def test_successful_type_compatibility_logging(caplog: Any) -> None:
assert "Compatible type pairs" in caplog.text
def test_complex_cycle_detection(caplog: Any) -> None:
caplog.set_level(logging.WARNING)
# Create a more complex graph with multiple cycles
executor1 = StringExecutor(id="executor1")
executor2 = StringExecutor(id="executor2")
executor3 = StringExecutor(id="executor3")
executor4 = StringExecutor(id="executor4")
# Create multiple paths and cycles
workflow = (
WorkflowBuilder()
.add_edge(executor1, executor2)
.add_edge(executor2, executor3)
.add_edge(executor3, executor4)
.add_edge(executor4, executor2) # Creates cycle: executor2 -> executor3 -> executor4 -> executor2
.set_start_executor(executor1)
.build()
)
assert workflow is not None
assert "Cycle detected in the workflow graph" in caplog.text
def test_no_cycles_in_simple_chain(caplog: Any) -> None:
caplog.set_level(logging.WARNING)
executor1 = StringExecutor(id="executor1")
executor2 = StringExecutor(id="executor2")
executor3 = StringExecutor(id="executor3")
# Simple chain without cycles
workflow = (
WorkflowBuilder()
.add_edge(executor1, executor2)
.add_edge(executor2, executor3)
.set_start_executor(executor1)
.build()
)
assert workflow is not None
# Should not log cycle detection
assert "Cycle detected" not in caplog.text
def test_multiple_dead_ends_detection(caplog: Any) -> None:
caplog.set_level(logging.INFO)
@@ -185,65 +185,6 @@ async def test_workflow_run_not_completed():
await workflow.run(NumberMessage(data=0))
async def test_workflow_send_responses_streaming():
"""Test the workflow run with approval."""
executor_a = IncrementExecutor(id="executor_a")
executor_b = MockExecutorRequestApproval(id="executor_b")
workflow = (
WorkflowBuilder()
.set_start_executor(executor_a)
.add_edge(executor_a, executor_b)
.add_edge(executor_b, executor_a)
.build()
)
request_info_event: RequestInfoEvent | None = None
async for event in workflow.run_stream(NumberMessage(data=0)):
if isinstance(event, RequestInfoEvent):
request_info_event = event
assert request_info_event is not None
result: int | None = None
completed = False
async for event in workflow.send_responses_streaming({
request_info_event.request_id: ApprovalMessage(approved=True)
}):
if isinstance(event, WorkflowOutputEvent):
result = event.data
elif isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
completed = True
assert (
completed and result is not None and result == 1
) # The data should be incremented by 1 from the initial message
async def test_workflow_send_responses():
"""Test the workflow run with approval."""
executor_a = IncrementExecutor(id="executor_a")
executor_b = MockExecutorRequestApproval(id="executor_b")
workflow = (
WorkflowBuilder()
.set_start_executor(executor_a)
.add_edge(executor_a, executor_b)
.add_edge(executor_b, executor_a)
.build()
)
events = await workflow.run(NumberMessage(data=0))
request_info_events = events.get_request_info_events()
assert len(request_info_events) == 1
result = await workflow.send_responses({request_info_events[0].request_id: ApprovalMessage(approved=True)})
assert result.get_final_state() == WorkflowRunState.IDLE
outputs = result.get_outputs()
assert outputs[0] == 1 # The data should be incremented by 1 from the initial message
async def test_fan_out():
"""Test a fan-out workflow."""
executor_a = IncrementExecutor(id="executor_a")
@@ -354,7 +295,7 @@ async def test_workflow_checkpointing_not_enabled_for_external_restore(simple_ex
# Attempt to restore from checkpoint without providing external storage should fail
try:
[event async for event in workflow.run_stream_from_checkpoint("fake-checkpoint-id")]
[event async for event in workflow.run_stream(checkpoint_id="fake-checkpoint-id")]
raise AssertionError("Expected ValueError to be raised")
except ValueError as e:
assert "Cannot restore from checkpoint" in str(e)
@@ -372,7 +313,7 @@ async def test_workflow_run_stream_from_checkpoint_no_checkpointing_enabled(simp
# Attempt to run from checkpoint should fail
try:
async for _ in workflow.run_stream_from_checkpoint("fake_checkpoint_id"):
async for _ in workflow.run_stream(checkpoint_id="fake_checkpoint_id"):
pass
raise AssertionError("Expected ValueError to be raised")
except ValueError as e:
@@ -396,7 +337,7 @@ async def test_workflow_run_stream_from_checkpoint_invalid_checkpoint(simple_exe
# Attempt to run from non-existent checkpoint should fail
try:
async for _ in workflow.run_stream_from_checkpoint("nonexistent_checkpoint_id"):
async for _ in workflow.run_stream(checkpoint_id="nonexistent_checkpoint_id"):
pass
raise AssertionError("Expected RuntimeError to be raised")
except RuntimeError as e:
@@ -427,8 +368,8 @@ async def test_workflow_run_stream_from_checkpoint_with_external_storage(simple_
# Resume from checkpoint using external storage parameter
try:
events: list[WorkflowEvent] = []
async for event in workflow_without_checkpointing.run_stream_from_checkpoint(
checkpoint_id, checkpoint_storage=storage
async for event in workflow_without_checkpointing.run_stream(
checkpoint_id=checkpoint_id, checkpoint_storage=storage
):
events.append(event)
if len(events) >= 2: # Limit to avoid infinite loops
@@ -463,14 +404,14 @@ async def test_workflow_run_from_checkpoint_non_streaming(simple_executor: Execu
.build()
)
# Test non-streaming run_from_checkpoint method
result = await workflow.run_from_checkpoint(checkpoint_id)
# Test non-streaming run method with checkpoint_id
result = await workflow.run(checkpoint_id=checkpoint_id)
assert isinstance(result, list) # Should return WorkflowRunResult which extends list
assert hasattr(result, "get_outputs") # Should have WorkflowRunResult methods
async def test_workflow_run_stream_from_checkpoint_with_responses(simple_executor: Executor):
"""Test that run_stream_from_checkpoint accepts responses parameter."""
"""Test that workflow can be resumed from checkpoint with pending RequestInfoEvents."""
with tempfile.TemporaryDirectory() as temp_dir:
storage = FileCheckpointStorage(temp_dir)
@@ -502,20 +443,16 @@ async def test_workflow_run_stream_from_checkpoint_with_responses(simple_executo
.build()
)
# Test that run_stream_from_checkpoint accepts responses parameter
responses = {"request_123": "test_response"}
# Resume from checkpoint - pending request events should be emitted
events: list[WorkflowEvent] = []
async for event in workflow.run_stream_from_checkpoint(checkpoint_id):
async for event in workflow.run_stream(checkpoint_id=checkpoint_id):
events.append(event)
# Verify that the pending request event was emitted
assert next(
event for event in events if isinstance(event, RequestInfoEvent) and event.request_id == "request_123"
)
async for event in workflow.send_responses_streaming(responses):
events.append(event)
assert len(events) > 0 # Just ensure we processed some events
@@ -594,6 +531,74 @@ async def test_workflow_multiple_runs_no_state_collision():
assert outputs1[0] != outputs3[0]
async def test_workflow_checkpoint_runtime_only_configuration(simple_executor: Executor):
"""Test that checkpointing can be configured ONLY at runtime, not at build time."""
with tempfile.TemporaryDirectory() as temp_dir:
storage = FileCheckpointStorage(temp_dir)
# Build workflow WITHOUT checkpointing at build time
workflow = (
WorkflowBuilder().add_edge(simple_executor, simple_executor).set_start_executor(simple_executor).build()
)
# Run with runtime checkpoint storage - should create checkpoints
test_message = Message(data="runtime checkpoint test", source_id="test", target_id=None)
result = await workflow.run(test_message, checkpoint_storage=storage)
assert result is not None
assert result.get_final_state() == WorkflowRunState.IDLE
# Verify checkpoints were created
checkpoints = await storage.list_checkpoints()
assert len(checkpoints) > 0
# Find a superstep checkpoint to resume from
checkpoints.sort(key=lambda cp: cp.timestamp)
resume_checkpoint = next(
(cp for cp in checkpoints if (cp.metadata or {}).get("checkpoint_type") == "superstep"),
checkpoints[-1],
)
# Create new workflow instance (still without build-time checkpointing)
workflow_resume = (
WorkflowBuilder().add_edge(simple_executor, simple_executor).set_start_executor(simple_executor).build()
)
# Resume from checkpoint using runtime checkpoint storage
result_resumed = await workflow_resume.run(
checkpoint_id=resume_checkpoint.checkpoint_id, checkpoint_storage=storage
)
assert result_resumed is not None
assert result_resumed.get_final_state() in (WorkflowRunState.IDLE, WorkflowRunState.IDLE_WITH_PENDING_REQUESTS)
async def test_workflow_checkpoint_runtime_overrides_buildtime(simple_executor: Executor):
"""Test that runtime checkpoint storage overrides build-time configuration."""
with tempfile.TemporaryDirectory() as temp_dir1, tempfile.TemporaryDirectory() as temp_dir2:
buildtime_storage = FileCheckpointStorage(temp_dir1)
runtime_storage = FileCheckpointStorage(temp_dir2)
# Build workflow with build-time checkpointing
workflow = (
WorkflowBuilder()
.add_edge(simple_executor, simple_executor)
.set_start_executor(simple_executor)
.with_checkpointing(buildtime_storage)
.build()
)
# Run with runtime checkpoint storage override
test_message = Message(data="override test", source_id="test", target_id=None)
result = await workflow.run(test_message, checkpoint_storage=runtime_storage)
assert result is not None
# Verify checkpoints were created in runtime storage, not build-time storage
buildtime_checkpoints = await buildtime_storage.list_checkpoints()
runtime_checkpoints = await runtime_storage.list_checkpoints()
assert len(runtime_checkpoints) > 0, "Runtime storage should have checkpoints"
assert len(buildtime_checkpoints) == 0, "Build-time storage should have no checkpoints when overridden"
async def test_comprehensive_edge_groups_workflow():
"""Test a workflow that uses SwitchCaseEdgeGroup, FanOutEdgeGroup, and FanInEdgeGroup."""
from agent_framework import Case, Default
@@ -799,9 +804,6 @@ async def test_workflow_concurrent_execution_prevention_mixed_methods():
async for _ in workflow.run_stream(NumberMessage(data=0)):
break
with pytest.raises(RuntimeError, match="Workflow is already running. Concurrent executions are not allowed."):
await workflow.send_responses({"test": "data"})
# Wait for the original task to complete
await task1
@@ -884,3 +886,48 @@ async def test_agent_streaming_vs_non_streaming() -> None:
if e.data and e.data.contents and e.data.contents[0].text
)
assert accumulated_text == "Hello World", f"Expected 'Hello World', got '{accumulated_text}'"
async def test_workflow_run_parameter_validation(simple_executor: Executor) -> None:
"""Test that run() and run_stream() properly validate parameter combinations."""
workflow = WorkflowBuilder().add_edge(simple_executor, simple_executor).set_start_executor(simple_executor).build()
test_message = Message(data="test", source_id="test", target_id=None)
# Valid: message only (new run)
result = await workflow.run(test_message)
assert result.get_final_state() == WorkflowRunState.IDLE
# Invalid: both message and checkpoint_id
with pytest.raises(ValueError, match="Cannot provide both 'message' and 'checkpoint_id'"):
await workflow.run(test_message, checkpoint_id="fake_id")
# Invalid: both message and checkpoint_id (streaming)
with pytest.raises(ValueError, match="Cannot provide both 'message' and 'checkpoint_id'"):
async for _ in workflow.run_stream(test_message, checkpoint_id="fake_id"):
pass
# Invalid: none of message or checkpoint_id
with pytest.raises(ValueError, match="Must provide either"):
await workflow.run()
# Invalid: none of message or checkpoint_id (streaming)
with pytest.raises(ValueError, match="Must provide either"):
async for _ in workflow.run_stream():
pass
async def test_workflow_run_stream_parameter_validation(simple_executor: Executor) -> None:
"""Test run_stream() specific parameter validation scenarios."""
workflow = WorkflowBuilder().add_edge(simple_executor, simple_executor).set_start_executor(simple_executor).build()
test_message = Message(data="test", source_id="test", target_id=None)
# Valid: message only (new run)
events: list[WorkflowEvent] = []
async for event in workflow.run_stream(test_message):
events.append(event)
assert any(isinstance(e, WorkflowStatusEvent) and e.state == WorkflowRunState.IDLE for e in events)
# Invalid combinations already tested in test_workflow_run_parameter_validation
# This test ensures streaming works correctly for valid parameters
+1 -1
View File
@@ -4,7 +4,7 @@ description = "Debug UI for Microsoft Agent Framework with OpenAI-compatible API
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
readme = "README.md"
requires-python = ">=3.10"
version = "1.0.0b251028"
version = "1.0.0b251104"
license-files = ["LICENSE"]
urls.homepage = "https://github.com/microsoft/agent-framework"
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
+1 -1
View File
@@ -4,7 +4,7 @@ description = "Experimental modules for Microsoft Agent Framework"
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
readme = "README.md"
requires-python = ">=3.10"
version = "1.0.0b251028"
version = "1.0.0b251104"
license-files = ["LICENSE"]
urls.homepage = "https://aka.ms/agent-framework"
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
+1 -1
View File
@@ -4,7 +4,7 @@ description = "Mem0 integration for Microsoft Agent Framework."
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
readme = "README.md"
requires-python = ">=3.10"
version = "1.0.0b251028"
version = "1.0.0b251104"
license-files = ["LICENSE"]
urls.homepage = "https://aka.ms/agent-framework"
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
+1 -1
View File
@@ -4,7 +4,7 @@ description = "Microsoft Purview (Graph dataSecurityAndGovernance) integration f
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
readme = "README.md"
requires-python = ">=3.10"
version = "1.0.0b251028"
version = "1.0.0b251104"
license-files = ["LICENSE"]
urls.homepage = "https://github.com/microsoft/agent-framework"
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
+1 -1
View File
@@ -4,7 +4,7 @@ description = "Redis integration for Microsoft Agent Framework."
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
readme = "README.md"
requires-python = ">=3.10"
version = "1.0.0b251028"
version = "1.0.0b251104"
license-files = ["LICENSE"]
urls.homepage = "https://aka.ms/agent-framework"
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
+5 -3
View File
@@ -4,7 +4,7 @@ description = "Microsoft Agent Framework for building AI Agents with Python. Thi
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
readme = "README.md"
requires-python = ">=3.10"
version = "1.0.0b251028"
version = "1.0.0b251104"
license-files = ["LICENSE"]
urls.homepage = "https://aka.ms/agent-framework"
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -24,13 +24,14 @@ classifiers = [
dependencies = [
"agent-framework-core",
"agent-framework-a2a",
"agent-framework-anthropic",
"agent-framework-azure-ai",
"agent-framework-copilotstudio",
"agent-framework-devui",
"agent-framework-lab",
"agent-framework-mem0",
"agent-framework-redis",
"agent-framework-devui",
"agent-framework-purview",
"agent-framework-redis",
]
[dependency-groups]
@@ -94,6 +95,7 @@ agent-framework-mem0 = { workspace = true }
agent-framework-redis = { workspace = true }
agent-framework-devui = { workspace = true }
agent-framework-purview = { workspace = true }
agent-framework-anthropic = { workspace = true }
[tool.ruff]
line-length = 120
+2 -1
View File
@@ -14,7 +14,8 @@ This directory contains samples demonstrating the capabilities of Microsoft Agen
| File | Description |
|------|-------------|
| [`getting_started/agents/anthropic/anthropic_with_openai_chat_client.py`](./getting_started/agents/anthropic/anthropic_with_openai_chat_client.py) | Anthropic with OpenAI Chat Client Example |
| [`getting_started/agents/anthropic/anthropic_basic.py`](./getting_started/agents/anthropic/anthropic_basic.py) | Agent with Anthropic Client |
| [`getting_started/agents/anthropic/anthropic_advanced.py`](./getting_started/agents/anthropic/anthropic_advanced.py) | Advanced sample with `thinking` and hosted tools. |
### Azure AI
@@ -6,12 +6,12 @@ This folder contains examples demonstrating how to use Anthropic's Claude models
| File | Description |
|------|-------------|
| [`anthropic_with_openai_chat_client.py`](anthropic_with_openai_chat_client.py) | Demonstrates how to configure OpenAI Chat Client to use Anthropic's Claude models. Shows both streaming and non-streaming responses with tool calling capabilities. |
| [`anthropic_basic.py`](anthropic_basic.py) | Demonstrates how to setup a simple agent using the AnthropicClient, with both streaming and non-streaming responses. |
| [`anthropic_advanced.py`](anthropic_advanced.py) | Shows advanced usage of the AnthropicClient, including hosted tools and `thinking`. |
## Environment Variables
Set the following environment variables before running the examples:
- `ANTHROPIC_API_KEY`: Your Anthropic API key (get one from [Anthropic Console](https://console.anthropic.com/))
- `ANTHROPIC_MODEL`: The Claude model to use (e.g., `claude-3-5-sonnet-20241022`, `claude-3-haiku-20240307`)
- `ANTHROPIC_MODEL`: The Claude model to use (e.g., `claude-haiku-4-5`, `claude-sonnet-4-5-20250929`)
@@ -0,0 +1,58 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
from agent_framework import HostedMCPTool, HostedWebSearchTool, TextReasoningContent, UsageContent
from agent_framework.anthropic import AnthropicClient
"""
Anthropic Chat Agent Example
This sample demonstrates using Anthropic with:
- Setting up an Anthropic-based agent with hosted tools.
- Using the `thinking` feature.
- Displaying both thinking and usage information during streaming responses.
"""
async def streaming_example() -> None:
"""Example of streaming response (get results as they are generated)."""
agent = AnthropicClient().create_agent(
name="DocsAgent",
instructions="You are a helpful agent for both Microsoft docs questions and general questions.",
tools=[
HostedMCPTool(
name="Microsoft Learn MCP",
url="https://learn.microsoft.com/api/mcp",
),
HostedWebSearchTool(),
],
# anthropic needs a value for the max_tokens parameter
# we set it to 1024, but you can override like this:
max_tokens=20000,
additional_chat_options={"thinking": {"type": "enabled", "budget_tokens": 10000}},
)
query = "Can you compare Python decorators with C# attributes?"
print(f"User: {query}")
print("Agent: ", end="", flush=True)
async for chunk in agent.run_stream(query):
for content in chunk.contents:
if isinstance(content, TextReasoningContent):
print(f"\033[32m{content.text}\033[0m", end="", flush=True)
if isinstance(content, UsageContent):
print(f"\n\033[34m[Usage so far: {content.details}]\033[0m\n", end="", flush=True)
if chunk.text:
print(chunk.text, end="", flush=True)
print("\n")
async def main() -> None:
print("=== Anthropic Example ===")
await streaming_example()
if __name__ == "__main__":
asyncio.run(main())
@@ -1,17 +1,15 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from random import randint
from typing import Annotated
from agent_framework.openai import OpenAIChatClient
from agent_framework.anthropic import AnthropicClient
"""
Anthropic with OpenAI Chat Client Example
Anthropic Chat Agent Example
This sample demonstrates using Anthropic models through OpenAI Chat Client by
configuring the base URL to point to Anthropic's API for cross-provider compatibility.
This sample demonstrates using Anthropic with an agent and a single custom tool.
"""
@@ -27,10 +25,7 @@ async def non_streaming_example() -> None:
"""Example of non-streaming response (get the complete result at once)."""
print("=== Non-streaming Response Example ===")
agent = OpenAIChatClient(
api_key=os.getenv("ANTHROPIC_API_KEY"),
base_url="https://api.anthropic.com/v1/",
model_id=os.getenv("ANTHROPIC_MODEL"),
agent = AnthropicClient(
).create_agent(
name="WeatherAgent",
instructions="You are a helpful weather agent.",
@@ -47,17 +42,14 @@ async def streaming_example() -> None:
"""Example of streaming response (get results as they are generated)."""
print("=== Streaming Response Example ===")
agent = OpenAIChatClient(
api_key=os.getenv("ANTHROPIC_API_KEY"),
base_url="https://api.anthropic.com/v1/",
model_id=os.getenv("ANTHROPIC_MODEL"),
agent = AnthropicClient(
).create_agent(
name="WeatherAgent",
instructions="You are a helpful weather agent.",
tools=get_weather,
)
query = "What's the weather like in Portland?"
query = "What's the weather like in Portland and in Paris?"
print(f"User: {query}")
print("Agent: ", end="", flush=True)
async for chunk in agent.run_stream(query):
@@ -67,10 +59,10 @@ async def streaming_example() -> None:
async def main() -> None:
print("=== Anthropic with OpenAI Chat Client Agent Example ===")
print("=== Anthropic Example ===")
await non_streaming_example()
await streaming_example()
await non_streaming_example()
if __name__ == "__main__":
@@ -5,14 +5,8 @@ from collections.abc import Collection
from typing import Any
from agent_framework import ChatMessage, ChatMessageStoreProtocol
from agent_framework._threads import ChatMessageStoreState
from agent_framework.openai import OpenAIChatClient
from pydantic import BaseModel
class CustomStoreState(BaseModel):
"""Implementation of custom chat message store state."""
messages: list[ChatMessage]
class CustomChatMessageStore(ChatMessageStoreProtocol):
@@ -32,13 +26,13 @@ class CustomChatMessageStore(ChatMessageStoreProtocol):
async def deserialize_state(self, serialized_store_state: Any, **kwargs: Any) -> None:
if serialized_store_state:
state = CustomStoreState.model_validate(serialized_store_state, **kwargs)
state = ChatMessageStoreState.from_dict(serialized_store_state, **kwargs)
if state.messages:
self._messages.extend(state.messages)
async def serialize_state(self, **kwargs: Any) -> Any:
state = CustomStoreState(messages=self._messages)
return state.model_dump(**kwargs)
state = ChatMessageStoreState(messages=self._messages)
return state.to_dict(**kwargs)
async def main() -> None:
@@ -117,14 +117,14 @@ async def main():
# retrieves the outputs yielded by any terminal nodes.
events = await workflow.run("hello world")
print(events.get_outputs())
# Summarize the final run state (e.g., COMPLETED)
# Summarize the final run state (e.g., IDLE)
print("Final state:", events.get_final_state())
"""
Sample Output:
['DLROW OLLEH']
Final state: WorkflowRunState.COMPLETED
Final state: WorkflowRunState.IDLE
"""
@@ -261,17 +261,21 @@ async def main() -> None:
pending_responses: dict[str, str] | None = None
completed = False
initial_run = True
while not completed:
last_executor: str | None = None
stream = (
workflow.send_responses_streaming(pending_responses)
if pending_responses is not None
else workflow.run_stream(
if initial_run:
stream = workflow.run_stream(
"Create a short launch blurb for the LumenX desk lamp. Emphasize adjustability and warm lighting."
)
)
pending_responses = None
initial_run = False
elif pending_responses is not None:
stream = workflow.send_responses_streaming(pending_responses)
pending_responses = None
else:
break
requests: list[tuple[str, DraftFeedbackRequest]] = []
async for event in stream:
@@ -250,7 +250,7 @@ async def run_interactive_session(
event_stream = workflow.run_stream(initial_message)
elif checkpoint_id:
print("\nStarting workflow from checkpoint...\n")
event_stream = workflow.run_stream_from_checkpoint(checkpoint_id)
event_stream = workflow.run_stream(checkpoint_id)
else:
raise ValueError("Either initial_message or checkpoint_id must be provided")
@@ -47,7 +47,7 @@ What you learn:
- How to configure FileCheckpointStorage and call with_checkpointing on WorkflowBuilder.
- How to list and inspect checkpoints programmatically.
- How to interactively choose a checkpoint to resume from (instead of always resuming
from the most recent or a hard-coded one) using run_stream_from_checkpoint.
from the most recent or a hard-coded one) using run_stream.
- How workflows complete by yielding outputs when idle, not via explicit completion events.
Prerequisites:
@@ -281,7 +281,7 @@ async def main():
new_workflow = create_workflow(checkpoint_storage=checkpoint_storage)
print(f"\nResuming from checkpoint: {chosen_cp_id}")
async for event in new_workflow.run_stream_from_checkpoint(chosen_cp_id, checkpoint_storage=checkpoint_storage):
async for event in new_workflow.run_stream(checkpoint_id=chosen_cp_id, checkpoint_storage=checkpoint_storage):
print(f"Resumed Event: {event}")
"""
@@ -356,7 +356,7 @@ async def main() -> None:
workflow2 = build_parent_workflow(storage)
request_info_event: RequestInfoEvent | None = None
async for event in workflow2.run_stream_from_checkpoint(
async for event in workflow2.run_stream(
resume_checkpoint.checkpoint_id,
):
if isinstance(event, RequestInfoEvent):
@@ -37,7 +37,7 @@ Show how to integrate a human step in the middle of an LLM workflow by using
Demonstrate:
- Alternating turns between an AgentExecutor and a human, driven by events.
- Using Pydantic response_format to enforce structured JSON output from the agent instead of regex parsing.
- Driving the loop in application code with run_stream and send_responses_streaming.
- Driving the loop in application code with run_stream and responses parameter.
Prerequisites:
- Azure OpenAI configured for AzureOpenAIChatClient with required environment variables.
@@ -32,7 +32,7 @@ Concepts highlighted here:
must keep stable IDs so the checkpoint state aligns when we rebuild the graph.
2. **Executor snapshotting** - checkpoints capture the pending plan-review request
map, at superstep boundaries.
3. **Resume with responses** - `Workflow.run_stream_from_checkpoint` accepts a
3. **Resume with responses** - `Workflow.send_responses_streaming` accepts a
`responses` mapping so we can inject the stored human reply during restoration.
Prerequisites:
@@ -141,7 +141,7 @@ async def main() -> None:
# Resume execution and capture the re-emitted plan review request.
request_info_event: RequestInfoEvent | None = None
async for event in resumed_workflow.run_stream_from_checkpoint(resume_checkpoint.checkpoint_id):
async for event in resumed_workflow.run_stream(checkpoint_id=resume_checkpoint.checkpoint_id):
if isinstance(event, RequestInfoEvent) and isinstance(event.data, MagenticPlanReviewRequest):
request_info_event = event
@@ -212,7 +212,7 @@ async def main() -> None:
final_event_post: WorkflowOutputEvent | None = None
post_emitted_events = False
post_plan_workflow = build_workflow(checkpoint_storage)
async for event in post_plan_workflow.run_stream_from_checkpoint(post_plan_checkpoint.checkpoint_id):
async for event in post_plan_workflow.run_stream(checkpoint_id=post_plan_checkpoint.checkpoint_id):
post_emitted_events = True
if isinstance(event, WorkflowOutputEvent):
final_event_post = event
+3486 -3439
View File
File diff suppressed because it is too large Load Diff