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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion src/claude_agent_sdk/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@
from dataclasses import dataclass
from typing import Any, Generic, TypeVar

from mcp.types import ToolAnnotations

from ._errors import (
ClaudeSDKError,
CLIConnectionError,
Expand Down Expand Up @@ -78,10 +80,14 @@ class SdkMcpTool(Generic[T]):
description: str
input_schema: type[T] | dict[str, Any]
handler: Callable[[T], Awaitable[dict[str, Any]]]
annotations: ToolAnnotations | None = None


def tool(
name: str, description: str, input_schema: type | dict[str, Any]
name: str,
description: str,
input_schema: type | dict[str, Any],
annotations: ToolAnnotations | None = None,
) -> Callable[[Callable[[Any], Awaitable[dict[str, Any]]]], SdkMcpTool[Any]]:
"""Decorator for defining MCP tools with type safety.

Expand Down Expand Up @@ -138,6 +144,7 @@ def decorator(
description=description,
input_schema=input_schema,
handler=handler,
annotations=annotations,
)

return decorator
Expand Down Expand Up @@ -268,6 +275,7 @@ async def list_tools() -> list[Tool]:
name=tool_def.name,
description=tool_def.description,
inputSchema=schema,
annotations=tool_def.annotations,
)
)
return tool_list
Expand Down Expand Up @@ -372,6 +380,7 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> Any:
"create_sdk_mcp_server",
"tool",
"SdkMcpTool",
"ToolAnnotations",
# Errors
"ClaudeSDKError",
"CLIConnectionError",
Expand Down
12 changes: 8 additions & 4 deletions src/claude_agent_sdk/_internal/query.py
Original file line number Diff line number Diff line change
Expand Up @@ -443,8 +443,9 @@ async def _handle_sdk_mcp_request(
if handler:
result = await handler(request)
# Convert MCP result to JSONRPC response
tools_data = [
{
tools_data = []
for tool in result.root.tools: # type: ignore[union-attr]
tool_data: dict[str, Any] = {
"name": tool.name,
"description": tool.description,
"inputSchema": (
Expand All @@ -455,8 +456,11 @@ async def _handle_sdk_mcp_request(
if tool.inputSchema
else {},
}
for tool in result.root.tools # type: ignore[union-attr]
]
if tool.annotations:
tool_data["annotations"] = tool.annotations.model_dump(
exclude_none=True
)
tools_data.append(tool_data)
return {
"jsonrpc": "2.0",
"id": message.get("id"),
Expand Down
116 changes: 116 additions & 0 deletions tests/test_sdk_mcp_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@

from claude_agent_sdk import (
ClaudeAgentOptions,
ToolAnnotations,
create_sdk_mcp_server,
tool,
)
Expand Down Expand Up @@ -263,3 +264,118 @@ async def generate_chart(args: dict[str, Any]) -> dict[str, Any]:
assert len(tool_executions) == 1
assert tool_executions[0]["name"] == "generate_chart"
assert tool_executions[0]["args"]["title"] == "Sales Report"


@pytest.mark.asyncio
async def test_tool_annotations():
"""Test that tool annotations are stored and flow through list_tools."""

@tool(
"read_data",
"Read data from source",
{"source": str},
annotations=ToolAnnotations(readOnlyHint=True),
)
async def read_data(args: dict[str, Any]) -> dict[str, Any]:
return {"content": [{"type": "text", "text": f"Data from {args['source']}"}]}

@tool(
"delete_item",
"Delete an item",
{"id": str},
annotations=ToolAnnotations(destructiveHint=True, idempotentHint=True),
)
async def delete_item(args: dict[str, Any]) -> dict[str, Any]:
return {"content": [{"type": "text", "text": f"Deleted {args['id']}"}]}

@tool(
"search",
"Search the web",
{"query": str},
annotations=ToolAnnotations(openWorldHint=True),
)
async def search(args: dict[str, Any]) -> dict[str, Any]:
return {"content": [{"type": "text", "text": f"Results for {args['query']}"}]}

@tool("no_annotations", "Tool without annotations", {"x": str})
async def no_annotations(args: dict[str, Any]) -> dict[str, Any]:
return {"content": [{"type": "text", "text": args["x"]}]}

# Verify annotations stored on SdkMcpTool
assert read_data.annotations is not None
assert read_data.annotations.readOnlyHint is True
assert delete_item.annotations is not None
assert delete_item.annotations.destructiveHint is True
assert delete_item.annotations.idempotentHint is True
assert search.annotations is not None
assert search.annotations.openWorldHint is True
assert no_annotations.annotations is None

# Verify annotations flow through list_tools handler
server_config = create_sdk_mcp_server(
name="annotations-test",
tools=[read_data, delete_item, search, no_annotations],
)
server = server_config["instance"]

from mcp.types import ListToolsRequest

list_handler = server.request_handlers[ListToolsRequest]
request = ListToolsRequest(method="tools/list")
response = await list_handler(request)

tools_by_name = {t.name: t for t in response.root.tools}

assert tools_by_name["read_data"].annotations is not None
assert tools_by_name["read_data"].annotations.readOnlyHint is True
assert tools_by_name["delete_item"].annotations is not None
assert tools_by_name["delete_item"].annotations.destructiveHint is True
assert tools_by_name["delete_item"].annotations.idempotentHint is True
assert tools_by_name["search"].annotations is not None
assert tools_by_name["search"].annotations.openWorldHint is True
assert tools_by_name["no_annotations"].annotations is None


@pytest.mark.asyncio
async def test_tool_annotations_in_jsonrpc():
"""Test that annotations are included in JSONRPC tools/list response."""
from claude_agent_sdk._internal.query import Query

@tool(
"read_only_tool",
"A read-only tool",
{"input": str},
annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=False),
)
async def read_only_tool(args: dict[str, Any]) -> dict[str, Any]:
return {"content": [{"type": "text", "text": args["input"]}]}

@tool("plain_tool", "A tool without annotations", {"input": str})
async def plain_tool(args: dict[str, Any]) -> dict[str, Any]:
return {"content": [{"type": "text", "text": args["input"]}]}

server_config = create_sdk_mcp_server(
name="jsonrpc-annotations-test",
tools=[read_only_tool, plain_tool],
)

# Simulate the JSONRPC tools/list request
query_instance = Query.__new__(Query)
query_instance.sdk_mcp_servers = {"test": server_config["instance"]}

response = await query_instance._handle_sdk_mcp_request(
"test",
{"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}},
)

assert response is not None
tools_data = response["result"]["tools"]
tools_by_name = {t["name"]: t for t in tools_data}

# Tool with annotations should include them
assert "annotations" in tools_by_name["read_only_tool"]
assert tools_by_name["read_only_tool"]["annotations"]["readOnlyHint"] is True
assert tools_by_name["read_only_tool"]["annotations"]["openWorldHint"] is False

# Tool without annotations should not have the key
assert "annotations" not in tools_by_name["plain_tool"]