Skip to content
16 changes: 6 additions & 10 deletions python/packages/ag-ui/agent_framework_ag_ui/_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
from datetime import date, datetime
from typing import Any

from agent_framework import AgentResponseUpdate, ChatResponseUpdate, FunctionTool, ToolProtocol
from agent_framework import AgentResponseUpdate, ChatResponseUpdate, FunctionTool

# Role mapping constants
AGUI_TO_FRAMEWORK_ROLE: dict[str, str] = {
Expand Down Expand Up @@ -198,10 +198,10 @@ def convert_agui_tools_to_agent_framework(

def convert_tools_to_agui_format(
tools: (
ToolProtocol
FunctionTool
| Callable[..., Any]
| MutableMapping[str, Any]
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
| Sequence[FunctionTool | Callable[..., Any] | MutableMapping[str, Any]]
| None
),
) -> list[dict[str, Any]] | None:
Expand All @@ -223,7 +223,7 @@ def convert_tools_to_agui_format(

# Normalize to list
if not isinstance(tools, list):
tool_list: list[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]] = [tools] # type: ignore[list-item]
tool_list: list[FunctionTool | Callable[..., Any] | MutableMapping[str, Any]] = [tools] # type: ignore[list-item]
else:
tool_list = tools # type: ignore[assignment]

Expand Down Expand Up @@ -254,12 +254,8 @@ def convert_tools_to_agui_format(
"parameters": ai_func.parameters(),
}
)
elif isinstance(tool_item, ToolProtocol):
# Handle other ToolProtocol implementations
# For now, we'll skip non-FunctionTool instances as they may not have
# the parameters() method. This matches .NET behavior which only
# converts FunctionToolDeclaration instances.
continue
# Note: dict-based hosted tools (CodeInterpreter, WebSearch, etc.) are passed through
# as-is in the first branch. Non-FunctionTool, non-dict items are skipped.

return results if results else None

Expand Down
163 changes: 128 additions & 35 deletions python/packages/anthropic/agent_framework_anthropic/_chat_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,6 @@
ChatResponseUpdate,
Content,
FunctionTool,
HostedCodeInterpreterTool,
HostedMCPTool,
HostedWebSearchTool,
TextSpanRegion,
UsageDetails,
get_logger,
Expand Down Expand Up @@ -331,6 +328,109 @@ class MyOptions(AnthropicChatOptions, total=False):
# streaming requires tracking the last function call ID and name
self._last_call_id_name: tuple[str, str] | None = None

# region Static factory methods for hosted tools

@staticmethod
def get_code_interpreter_tool(
*,
type_name: str | None = None,
) -> dict[str, Any]:
"""Create a code interpreter tool configuration for Anthropic.

Keyword Args:
type_name: Override the tool type name. Defaults to "code_execution_20250825".

Returns:
A dict-based tool configuration ready to pass to ChatAgent.

Examples:
.. code-block:: python

from agent_framework.anthropic import AnthropicClient

tool = AnthropicClient.get_code_interpreter_tool()
agent = AnthropicClient().as_agent(tools=[tool])
"""
return {"type": type_name or "code_execution_20250825"}

@staticmethod
def get_web_search_tool(
*,
type_name: str | None = None,
) -> dict[str, Any]:
"""Create a web search tool configuration for Anthropic.

Keyword Args:
type_name: Override the tool type name. Defaults to "web_search_20250305".

Returns:
A dict-based tool configuration ready to pass to ChatAgent.

Examples:
.. code-block:: python

from agent_framework.anthropic import AnthropicClient

tool = AnthropicClient.get_web_search_tool()
agent = AnthropicClient().as_agent(tools=[tool])
"""
return {"type": type_name or "web_search_20250305"}

@staticmethod
def get_mcp_tool(
*,
name: str,
url: str,
allowed_tools: list[str] | None = None,
authorization_token: str | None = None,
) -> dict[str, Any]:
"""Create a hosted MCP tool configuration for Anthropic.

This configures an MCP (Model Context Protocol) server that will be called
by Anthropic's service. The tools from this MCP server are executed remotely
by Anthropic, not locally by your application.

Note:
For local MCP execution where your application calls the MCP server
directly, use the MCP client tools instead of this method.

Keyword Args:
name: A label/name for the MCP server.
url: The URL of the MCP server.
allowed_tools: List of tool names that are allowed to be used from this MCP server.
authorization_token: Authorization token for the MCP server (e.g., Bearer token).

Returns:
A dict-based tool configuration ready to pass to ChatAgent.

Examples:
.. code-block:: python

from agent_framework.anthropic import AnthropicClient

tool = AnthropicClient.get_mcp_tool(
name="GitHub",
url="https://api.githubcopilot.com/mcp/",
authorization_token="Bearer ghp_xxx",
)
agent = AnthropicClient().as_agent(tools=[tool])
"""
result: dict[str, Any] = {
"type": "mcp",
"server_label": name.replace(" ", "_"),
"server_url": url,
}

if allowed_tools:
result["allowed_tools"] = allowed_tools

if authorization_token:
result["headers"] = {"authorization": authorization_token}

return result

# endregion

# region Get response methods

@override
Expand Down Expand Up @@ -583,43 +683,36 @@ def _prepare_tools_for_anthropic(self, options: dict[str, Any]) -> dict[str, Any
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 FunctionTool():
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_execution",
}
tool_list.append(code_tool)
case HostedMCPTool():
if isinstance(tool, FunctionTool):
tool_list.append({
"type": "custom",
"name": tool.name,
"description": tool.description,
"input_schema": tool.parameters(),
})
elif isinstance(tool, MutableMapping):
# Handle dict-based tools from static factory methods
tool_dict = tool if isinstance(tool, dict) else dict(tool)

if tool_dict.get("type") == "mcp":
# MCP servers must be routed to separate mcp_servers parameter
server_def: dict[str, Any] = {
"type": "url",
"name": tool.name,
"url": str(tool.url),
"name": tool_dict.get("server_label", ""),
"url": tool_dict.get("server_url", ""),
}
if tool.allowed_tools:
server_def["tool_configuration"] = {"allowed_tools": list(tool.allowed_tools)}
if tool.headers and (auth := tool.headers.get("authorization")):
if allowed_tools := tool_dict.get("allowed_tools"):
server_def["tool_configuration"] = {"allowed_tools": list(allowed_tools)}
headers = tool_dict.get("headers")
if isinstance(headers, dict) and (auth := 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")
else:
# Pass through all other dict-based tools directly
# (e.g., web_search_20250305, code_execution_20250825)
tool_list.append(tool_dict)
else:
logger.debug(f"Ignoring unsupported tool type: {type(tool)} for now")

if tool_list:
result["tools"] = tool_list
Expand Down
44 changes: 18 additions & 26 deletions python/packages/anthropic/tests/test_anthropic_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,6 @@
ChatOptions,
ChatResponseUpdate,
Content,
HostedCodeInterpreterTool,
HostedMCPTool,
HostedWebSearchTool,
tool,
)
from agent_framework.exceptions import ServiceInitializationError
Expand Down Expand Up @@ -278,37 +275,35 @@ def get_weather(location: Annotated[str, Field(description="Location to get weat


def test_prepare_tools_for_anthropic_web_search(mock_anthropic_client: MagicMock) -> None:
"""Test converting HostedWebSearchTool to Anthropic format."""
"""Test converting web_search dict tool to Anthropic format."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
chat_options = ChatOptions(tools=[HostedWebSearchTool()])
chat_options = ChatOptions(tools=[AnthropicClient.get_web_search_tool()])

result = chat_client._prepare_tools_for_anthropic(chat_options)

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_prepare_tools_for_anthropic_code_interpreter(mock_anthropic_client: MagicMock) -> None:
"""Test converting HostedCodeInterpreterTool to Anthropic format."""
"""Test converting code_interpreter dict tool to Anthropic format."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
chat_options = ChatOptions(tools=[HostedCodeInterpreterTool()])
chat_options = ChatOptions(tools=[AnthropicClient.get_code_interpreter_tool()])

result = chat_client._prepare_tools_for_anthropic(chat_options)

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_execution"


def test_prepare_tools_for_anthropic_mcp_tool(mock_anthropic_client: MagicMock) -> None:
"""Test converting HostedMCPTool to Anthropic format."""
"""Test converting MCP dict tool to Anthropic format."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
chat_options = ChatOptions(tools=[HostedMCPTool(name="test-mcp", url="https://example.com/mcp")])
chat_options = ChatOptions(tools=[AnthropicClient.get_mcp_tool(name="test-mcp", url="https://example.com/mcp")])

result = chat_client._prepare_tools_for_anthropic(chat_options)

Expand All @@ -321,23 +316,21 @@ def test_prepare_tools_for_anthropic_mcp_tool(mock_anthropic_client: MagicMock)


def test_prepare_tools_for_anthropic_mcp_with_auth(mock_anthropic_client: MagicMock) -> None:
"""Test converting HostedMCPTool with authorization headers."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
chat_options = ChatOptions(
tools=[
HostedMCPTool(
name="test-mcp",
url="https://example.com/mcp",
headers={"authorization": "Bearer token123"},
)
]
"""Test converting MCP dict tool with authorization token."""
chat_client = create_test_anthropic_client(mock_anthropic_client)
# Use the static method with authorization_token
mcp_tool = AnthropicClient.get_mcp_tool(
name="test-mcp",
url="https://example.com/mcp",
authorization_token="Bearer token123",
)
chat_options = ChatOptions(tools=[mcp_tool])

result = chat_client._prepare_tools_for_anthropic(chat_options)

assert result is not None
assert "mcp_servers" in result
# The authorization header is converted to authorization_token
# The authorization_token should be passed through
assert "authorization_token" in result["mcp_servers"][0]
assert result["mcp_servers"][0]["authorization_token"] == "Bearer token123"

Expand Down Expand Up @@ -776,12 +769,11 @@ async def test_anthropic_client_integration_hosted_tools() -> None:

messages = [ChatMessage("user", ["What tools do you have available?"])]
tools = [
HostedWebSearchTool(),
HostedCodeInterpreterTool(),
HostedMCPTool(
AnthropicClient.get_web_search_tool(),
AnthropicClient.get_code_interpreter_tool(),
AnthropicClient.get_mcp_tool(
name="example-mcp",
url="https://learn.microsoft.com/api/mcp",
approval_mode="never_require",
),
]

Expand Down
Loading
Loading