Coverage for agentos/llm/providers/anthropic.py: 0%

1"""Anthropic Claude API Provider.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import os 

7import urllib.request 

8from typing import Optional 

9 

10from agentos.llm.base import ( 

11 LLMProvider, Message, MessageRole, CompletionResult, 

12 CompletionChoice, CompletionUsage, Tool, ToolCall, 

13) 

14 

15 

16class AnthropicProvider(LLMProvider): 

17 """Anthropic Claude API provider. 

18 

19 Requires: ANTHROPIC_API_KEY env var. 

20 """ 

21 

22 provider_name = "anthropic" 

23 API_URL = "https://api.anthropic.com/v1/messages" 

24 ANTHROPIC_VERSION = "2023-06-01" 

25 

26 def __init__(self, model: Optional[str] = None, api_key: Optional[str] = None): 

27 super().__init__(model=model or "claude-3-5-sonnet-20241022") 

28 self._api_key = api_key or os.getenv("ANTHROPIC_API_KEY", "") 

29 

30 def _messages_to_anthropic(self, messages: list[Message]) -> tuple[list[dict], Optional[str]]: 

31 """Convert to Anthropic format. Returns (messages, system_prompt).""" 

32 system = None 

33 result = [] 

34 for m in messages: 

35 if m.role == MessageRole.SYSTEM: 

36 system = m.content 

37 continue 

38 entry: dict = {"role": m.role.value} 

39 if m.content: 

40 entry["content"] = [{"type": "text", "text": m.content}] 

41 if m.tool_calls: 

42 # Anthropic: assistant content with tool_use blocks 

43 content_blocks = [] 

44 if m.content: 

45 content_blocks.append({"type": "text", "text": m.content}) 

46 for tc in m.tool_calls: 

47 content_blocks.append({ 

48 "type": "tool_use", 

49 "id": tc.id, 

50 "name": tc.name, 

51 "input": json.loads(tc.arguments), 

52 }) 

53 entry["content"] = content_blocks 

54 if m.tool_call_id: 

55 entry["role"] = "user" 

56 entry["content"] = [{ 

57 "type": "tool_result", 

58 "tool_use_id": m.tool_call_id, 

59 "content": m.content or "", 

60 }] 

61 result.append(entry) 

62 return result, system 

63 

64 def _tools_to_anthropic(self, tools: list[Tool]) -> list[dict]: 

65 result = [] 

66 for t in tools: 

67 schema = t.as_schema() 

68 result.append({ 

69 "name": schema["function"]["name"], 

70 "description": schema["function"]["description"], 

71 "input_schema": schema["function"]["parameters"], 

72 }) 

73 return result 

74 

75 def chat(self, messages: list[Message], **kwargs) -> CompletionResult: 

76 tools_param = kwargs.get("tools", []) 

77 api_messages, system = self._messages_to_anthropic(messages) 

78 

79 body: dict = { 

80 "model": self.model, 

81 "messages": api_messages, 

82 "max_tokens": 4096, 

83 "stream": False, 

84 } 

85 if system: 

86 body["system"] = system 

87 if tools_param: 

88 body["tools"] = self._tools_to_anthropic(tools_param) 

89 

90 req = urllib.request.Request( 

91 self.API_URL, 

92 data=json.dumps(body).encode("utf-8"), 

93 headers={ 

94 "Content-Type": "application/json", 

95 "x-api-key": self._api_key, 

96 "anthropic-version": self.ANTHROPIC_VERSION, 

97 }, 

98 method="POST", 

99 ) 

100 

101 with urllib.request.urlopen(req, timeout=120) as resp: 

102 data = json.loads(resp.read().decode("utf-8")) 

103 

104 # Parse response 

105 content_blocks = data.get("content", []) 

106 text_content = "" 

107 tool_calls = [] 

108 

109 for block in content_blocks: 

110 if block.get("type") == "text": 

111 text_content += block.get("text", "") 

112 elif block.get("type") == "tool_use": 

113 tool_calls.append(ToolCall( 

114 id=block.get("id", ""), 

115 name=block.get("name", ""), 

116 arguments=json.dumps(block.get("input", {})), 

117 )) 

118 

119 return CompletionResult( 

120 id=data.get("id", ""), 

121 model=data.get("model", self.model), 

122 choices=[CompletionChoice( 

123 index=0, 

124 message=Message( 

125 role=MessageRole.ASSISTANT, 

126 content=text_content, 

127 tool_calls=tool_calls if tool_calls else None, 

128 ), 

129 finish_reason=data.get("stop_reason", "end_turn"), 

130 )], 

131 usage=CompletionUsage( 

132 prompt_tokens=data.get("usage", {}).get("input_tokens", 0), 

133 completion_tokens=data.get("usage", {}).get("output_tokens", 0), 

134 total_tokens=( 

135 data.get("usage", {}).get("input_tokens", 0) + 

136 data.get("usage", {}).get("output_tokens", 0) 

137 ), 

138 ), 

139 ) 

140 

141 async def achat(self, messages: list[Message], **kwargs) -> CompletionResult: 

142 return self.chat(messages, **kwargs)