Coverage for src / agent / providers / anthropic / types.py: 77%

1# Copyright 2025-2026 Microsoft Corporation 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15""" 

16Type definitions and conversion utilities for Anthropic provider. 

17 

18This module provides functions to convert between agent-framework's message 

19formats and Anthropic Claude's expected formats. 

20""" 

21 

22from typing import Any 

23 

24from agent_framework import ( 

25 AIFunction, 

26 ChatMessage, 

27 FunctionCallContent, 

28 FunctionResultContent, 

29 TextContent, 

30) 

31 

32 

33def to_anthropic_messages( 

34 messages: list[ChatMessage], 

35) -> tuple[str | None, list[dict[str, Any]]]: 

36 """Convert agent-framework ChatMessages to Anthropic message format. 

37 

38 Anthropic handles system messages separately from the conversation, 

39 so this function extracts the system prompt and converts the rest. 

40 

41 Args: 

42 messages: List of ChatMessage from agent-framework 

43 

44 Returns: 

45 Tuple of (system_prompt, messages_list) 

46 

47 Example: 

48 >>> msgs = [ChatMessage(role="user", contents=[TextContent(text="Hello")])] 

49 >>> system, anthropic_msgs = to_anthropic_messages(msgs) 

50 >>> anthropic_msgs[0]["role"] 

51 'user' 

52 """ 

53 system_prompt: str | None = None 

54 anthropic_messages: list[dict[str, Any]] = [] 

55 

56 for message in messages: 

57 # Handle Role enum or string 

58 role_str = message.role.value if hasattr(message.role, "value") else str(message.role) 

59 

60 # Extract system prompt separately (Anthropic handles it differently) 

61 if role_str == "system": 

62 # Collect system message text 

63 system_texts = [] 

64 for item in message.contents if hasattr(message, "contents") else []: 

65 if isinstance(item, TextContent): 

66 system_texts.append(item.text) 

67 if system_texts: 

68 system_prompt = "\n".join(system_texts) 

69 continue 

70 

71 # Convert to Anthropic message format 

72 anthropic_msg = _convert_single_message(message, role_str) 

73 if anthropic_msg: 

74 anthropic_messages.append(anthropic_msg) 

75 

76 return system_prompt, anthropic_messages 

77 

78 

79def _convert_single_message(message: ChatMessage, role_str: str) -> dict[str, Any] | None: 

80 """Convert a single ChatMessage to Anthropic format. 

81 

82 Args: 

83 message: ChatMessage to convert 

84 role_str: Role as string 

85 

86 Returns: 

87 Anthropic message dict or None if empty 

88 """ 

89 # Map roles (agent-framework -> Anthropic) 

90 role_mapping = { 

91 "user": "user", 

92 "assistant": "assistant", 

93 "tool": "user", # Tool results come from user role in Anthropic 

94 } 

95 anthropic_role = role_mapping.get(role_str, "user") 

96 

97 # Build content list 

98 content: list[dict[str, Any]] = [] 

99 contents = message.contents if hasattr(message, "contents") else [] 

100 

101 for item in contents: 

102 if isinstance(item, TextContent): 

103 content.append({"type": "text", "text": item.text}) 

104 

105 elif isinstance(item, FunctionCallContent): 

106 # Function calls are tool_use blocks 

107 content.append( 

108 { 

109 "type": "tool_use", 

110 "id": item.call_id, 

111 "name": item.name, 

112 "input": item.arguments or {}, 

113 } 

114 ) 

115 

116 elif isinstance(item, FunctionResultContent): 

117 # Function results are tool_result blocks 

118 result_content: str 

119 if isinstance(item.result, str): 

120 result_content = item.result 

121 else: 

122 import json 

123 

124 result_content = json.dumps(item.result) 

125 

126 content.append( 

127 { 

128 "type": "tool_result", 

129 "tool_use_id": item.call_id, 

130 "content": result_content, 

131 } 

132 ) 

133 

134 if not content: 

135 # Try to get text directly from message 

136 text_value = getattr(message, "text", "") 

137 if text_value: 

138 content.append({"type": "text", "text": text_value}) 

139 

140 if not content: 

141 return None 

142 

143 return {"role": anthropic_role, "content": content} 

144 

145 

146def from_anthropic_message(response: Any) -> ChatMessage: 

147 """Convert Anthropic response to agent-framework ChatMessage. 

148 

149 Args: 

150 response: Response object from Anthropic API 

151 

152 Returns: 

153 ChatMessage for agent-framework 

154 

155 Example: 

156 >>> response = await client.messages.create(...) 

157 >>> msg = from_anthropic_message(response) 

158 >>> msg.role 

159 'assistant' 

160 """ 

161 content_items: list[TextContent | FunctionCallContent] = [] 

162 

163 # Anthropic response has a 'content' list with content blocks 

164 if hasattr(response, "content"): 

165 for block in response.content: 

166 block_type = getattr(block, "type", None) 

167 

168 if block_type == "text": 

169 text = getattr(block, "text", "") 

170 if text: 

171 content_items.append(TextContent(text=text)) 

172 

173 elif block_type == "tool_use": 

174 # Tool use block contains function call 

175 content_items.append( 

176 FunctionCallContent( 

177 call_id=getattr(block, "id", f"call_{id(block)}"), 

178 name=getattr(block, "name", ""), 

179 arguments=getattr(block, "input", {}), 

180 ) 

181 ) 

182 

183 # Ensure at least empty text content 

184 if not content_items: 

185 content_items = [TextContent(text="")] 

186 

187 return ChatMessage(role="assistant", contents=content_items) 

188 

189 

190def from_anthropic_stream_event(event: Any) -> tuple[str | None, FunctionCallContent | None]: 

191 """Extract content from Anthropic streaming event. 

192 

193 Args: 

194 event: Streaming event from Anthropic API 

195 

196 Returns: 

197 Tuple of (text_delta, function_call) - one or both may be None 

198 """ 

199 text_delta: str | None = None 

200 function_call: FunctionCallContent | None = None 

201 

202 event_type = getattr(event, "type", "") 

203 

204 if event_type == "content_block_delta": 

205 delta = getattr(event, "delta", None) 

206 if delta: 

207 delta_type = getattr(delta, "type", "") 

208 if delta_type == "text_delta": 

209 text_delta = getattr(delta, "text", "") 

210 elif delta_type == "input_json_delta": 

211 # Partial JSON for tool input - handled by caller accumulation 

212 pass 

213 

214 elif event_type == "content_block_start": 

215 content_block = getattr(event, "content_block", None) 

216 if content_block: 

217 block_type = getattr(content_block, "type", "") 

218 if block_type == "tool_use": 

219 # Tool use started - create placeholder 

220 function_call = FunctionCallContent( 

221 call_id=getattr(content_block, "id", f"call_{id(content_block)}"), 

222 name=getattr(content_block, "name", ""), 

223 arguments={}, 

224 ) 

225 

226 return text_delta, function_call 

227 

228 

229def to_anthropic_tools(tools: list[AIFunction]) -> list[dict[str, Any]]: 

230 """Convert agent-framework AIFunction tools to Anthropic tool format. 

231 

232 Args: 

233 tools: List of AIFunction objects from agent-framework 

234 

235 Returns: 

236 List of Anthropic tool definitions 

237 

238 Example: 

239 >>> tools = [AIFunction(name="get_weather", description="Get weather")] 

240 >>> anthropic_tools = to_anthropic_tools(tools) 

241 >>> anthropic_tools[0]["name"] 

242 'get_weather' 

243 """ 

244 anthropic_tools: list[dict[str, Any]] = [] 

245 

246 for tool in tools: 

247 tool_def: dict[str, Any] = { 

248 "name": tool.name, 

249 "description": tool.description or "", 

250 } 

251 

252 # Add input schema if available 

253 if hasattr(tool, "parameters"): 

254 params = tool.parameters() if callable(tool.parameters) else tool.parameters 

255 if params: 

256 tool_def["input_schema"] = _convert_parameters_to_schema(params) 

257 else: 

258 # Anthropic requires input_schema even if empty 

259 tool_def["input_schema"] = {"type": "object", "properties": {}} 

260 else: 

261 tool_def["input_schema"] = {"type": "object", "properties": {}} 

262 

263 anthropic_tools.append(tool_def) 

264 

265 return anthropic_tools 

266 

267 

268def _convert_parameters_to_schema(parameters: dict[str, Any]) -> dict[str, Any]: 

269 """Convert agent-framework parameter schema to Anthropic input_schema format. 

270 

271 Args: 

272 parameters: Parameter schema from AIFunction 

273 

274 Returns: 

275 Anthropic-compatible JSON schema 

276 """ 

277 schema: dict[str, Any] = { 

278 "type": "object", 

279 "properties": parameters.get("properties", {}), 

280 } 

281 

282 if "required" in parameters: 

283 schema["required"] = parameters["required"] 

284 

285 return schema 

286 

287 

288def extract_usage_metadata(response: Any) -> dict[str, Any]: 

289 """Extract token usage information from Anthropic response. 

290 

291 Args: 

292 response: Response object from Anthropic API 

293 

294 Returns: 

295 Dictionary with usage metadata 

296 

297 Example: 

298 >>> usage = extract_usage_metadata(response) 

299 >>> usage["prompt_tokens"] 

300 15 

301 """ 

302 usage: dict[str, Any] = {} 

303 

304 if hasattr(response, "usage"): 

305 response_usage = response.usage 

306 usage["prompt_tokens"] = getattr(response_usage, "input_tokens", 0) 

307 usage["completion_tokens"] = getattr(response_usage, "output_tokens", 0) 

308 usage["total_tokens"] = usage["prompt_tokens"] + usage["completion_tokens"] 

309 

310 return usage