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 

8 

9from agentos.llm.base import ( 

10 CompletionChoice, 

11 CompletionResult, 

12 CompletionUsage, 

13 LLMProvider, 

14 Message, 

15 MessageRole, 

16 Tool, 

17 ToolCall, 

18) 

19 

20 

21class AnthropicProvider(LLMProvider): 

22 """Anthropic Claude API provider. 

23 

24 Requires: ANTHROPIC_API_KEY env var. 

25 """ 

26 

27 provider_name = "anthropic" 

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

29 ANTHROPIC_VERSION = "2023-06-01" 

30 

31 def __init__(self, model: str | None = None, api_key: str | None = None): 

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

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

34 

35 def _messages_to_anthropic(self, messages: list[Message]) -> tuple[list[dict], str | None]: 

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

37 system = None 

38 result = [] 

39 for m in messages: 

40 if m.role == MessageRole.SYSTEM: 

41 system = m.content 

42 continue 

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

44 if m.content: 

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

46 if m.tool_calls: 

47 # Anthropic: assistant content with tool_use blocks 

48 content_blocks = [] 

49 if m.content: 

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

51 for tc in m.tool_calls: 

52 content_blocks.append( 

53 { 

54 "type": "tool_use", 

55 "id": tc.id, 

56 "name": tc.name, 

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

58 } 

59 ) 

60 entry["content"] = content_blocks 

61 if m.tool_call_id: 

62 entry["role"] = "user" 

63 entry["content"] = [ 

64 { 

65 "type": "tool_result", 

66 "tool_use_id": m.tool_call_id, 

67 "content": m.content or "", 

68 } 

69 ] 

70 result.append(entry) 

71 return result, system 

72 

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

74 result = [] 

75 for t in tools: 

76 schema = t.as_schema() 

77 result.append( 

78 { 

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

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

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

82 } 

83 ) 

84 return result 

85 

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

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

88 api_messages, system = self._messages_to_anthropic(messages) 

89 

90 body: dict = { 

91 "model": self.model, 

92 "messages": api_messages, 

93 "max_tokens": 4096, 

94 "stream": False, 

95 } 

96 if system: 

97 body["system"] = system 

98 if tools_param: 

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

100 

101 req = urllib.request.Request( 

102 self.API_URL, 

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

104 headers={ 

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

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

107 "anthropic-version": self.ANTHROPIC_VERSION, 

108 }, 

109 method="POST", 

110 ) 

111 

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

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

114 

115 # Parse response 

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

117 text_content = "" 

118 tool_calls = [] 

119 

120 for block in content_blocks: 

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

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

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

124 tool_calls.append( 

125 ToolCall( 

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

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

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

129 ) 

130 ) 

131 

132 return CompletionResult( 

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

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

135 choices=[ 

136 CompletionChoice( 

137 index=0, 

138 message=Message( 

139 role=MessageRole.ASSISTANT, 

140 content=text_content, 

141 tool_calls=tool_calls if tool_calls else None, 

142 ), 

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

144 ) 

145 ], 

146 usage=CompletionUsage( 

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

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

149 total_tokens=( 

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

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

152 ), 

153 ), 

154 ) 

155 

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

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