Coverage for agentos/models/backends/anthropic.py: 27%

1"""Anthropic Claude backend for AgentOS. 

2 

3Supports Claude 3.5 Sonnet, Claude 3 Opus, Claude 3 Haiku. 

4Uses Anthropic Messages API. 

5""" 

6 

7import os 

8from collections.abc import AsyncIterator 

9from dataclasses import dataclass, field 

10from typing import Any 

11 

12 

13@dataclass 

14class ClaudeConfig: 

15 """Configuration for Anthropic Claude backend.""" 

16 

17 api_key: str = field(default_factory=lambda: os.environ.get("ANTHROPIC_API_KEY", "")) 

18 base_url: str = "https://api.anthropic.com/v1" 

19 model: str = "claude-sonnet-4-20250514" 

20 temperature: float = 0.7 

21 max_tokens: int = 4096 

22 top_p: float = 1.0 

23 top_k: int = -1 

24 timeout: int = 90 

25 max_retries: int = 3 

26 anthropic_version: str = "2023-06-01" 

27 

28 

29class ClaudeClient: 

30 """Anthropic Claude LLM client. 

31 

32 Supports: 

33 - Claude Opus 4 / Claude Sonnet 4 / Claude Haiku 3.5 

34 - Streaming and non-streaming 

35 - Tool use (function calling) 

36 - System prompts 

37 """ 

38 

39 def __init__(self, config: ClaudeConfig | None = None): 

40 self.config = config or ClaudeConfig() 

41 

42 @property 

43 def _headers(self) -> dict[str, str]: 

44 return { 

45 "x-api-key": self.config.api_key, 

46 "anthropic-version": self.config.anthropic_version, 

47 "content-type": "application/json", 

48 } 

49 

50 @property 

51 def _messages_url(self) -> str: 

52 return f"{self.config.base_url}/messages" 

53 

54 def _build_payload( 

55 self, 

56 messages: list[dict[str, str]], 

57 system: str | None = None, 

58 **kwargs, 

59 ) -> dict[str, Any]: 

60 payload: dict[str, Any] = { 

61 "model": self.config.model, 

62 "max_tokens": kwargs.get("max_tokens", self.config.max_tokens), 

63 "messages": messages, 

64 } 

65 

66 if system: 

67 payload["system"] = system 

68 

69 temp = kwargs.get("temperature", self.config.temperature) 

70 if temp > 0: 

71 payload["temperature"] = temp 

72 

73 if kwargs.get("top_p", self.config.top_p) < 1.0: 

74 payload["top_p"] = kwargs.get("top_p", self.config.top_p) 

75 

76 if self.config.top_k > 0: 

77 payload["top_k"] = self.config.top_k 

78 

79 if kwargs.get("tools"): 

80 payload["tools"] = kwargs["tools"] 

81 

82 return payload 

83 

84 async def _async_request(self, payload: dict) -> dict: 

85 import httpx 

86 

87 timeout = httpx.Timeout(self.config.timeout) 

88 for attempt in range(self.config.max_retries): 

89 try: 

90 async with httpx.AsyncClient() as client: 

91 resp = await client.post( 

92 self._messages_url, 

93 json=payload, 

94 headers=self._headers, 

95 timeout=timeout, 

96 ) 

97 resp.raise_for_status() 

98 return resp.json() 

99 except httpx.HTTPStatusError as e: 

100 if attempt == self.config.max_retries - 1: 

101 raise 

102 if e.response.status_code >= 500 or e.response.status_code == 429: 

103 import asyncio 

104 

105 await asyncio.sleep(2**attempt) 

106 continue 

107 raise 

108 

109 async def chat( 

110 self, 

111 messages: list[dict[str, str]], 

112 system: str | None = None, 

113 **kwargs, 

114 ) -> dict[str, Any]: 

115 """Send a chat completion request. 

116 

117 Messages format: [{"role": "user", "content": "..."}, ...] 

118 Return format: {"content": str, "role": str, "usage": dict, "model": str} 

119 """ 

120 payload = self._build_payload(messages, system, **kwargs) 

121 result = await self._async_request(payload) 

122 

123 content_blocks = result.get("content", []) 

124 text_content = "" 

125 tool_calls = [] 

126 

127 for block in content_blocks: 

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

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

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

131 tool_calls.append( 

132 { 

133 "id": block.get("id", ""), 

134 "name": block.get("name", ""), 

135 "arguments": block.get("input", {}), 

136 } 

137 ) 

138 

139 usage = result.get("usage", {}) 

140 return { 

141 "content": text_content, 

142 "role": "assistant", 

143 "usage": { 

144 "input_tokens": usage.get("input_tokens", 0), 

145 "output_tokens": usage.get("output_tokens", 0), 

146 }, 

147 "model": result.get("model", self.config.model), 

148 "stop_reason": result.get("stop_reason", ""), 

149 "tool_calls": tool_calls, 

150 } 

151 

152 async def chat_stream( 

153 self, 

154 messages: list[dict[str, str]], 

155 system: str | None = None, 

156 **kwargs, 

157 ) -> AsyncIterator[dict[str, Any]]: 

158 """Stream chat completion events.""" 

159 payload = self._build_payload(messages, system, **kwargs) 

160 payload["stream"] = True 

161 

162 import httpx 

163 

164 timeout = httpx.Timeout(self.config.timeout * 2) 

165 async with httpx.AsyncClient() as client: 

166 async with client.stream( 

167 "POST", 

168 self._messages_url, 

169 json=payload, 

170 headers=self._headers, 

171 timeout=timeout, 

172 ) as response: 

173 response.raise_for_status() 

174 async for line in response.aiter_lines(): 

175 if line.startswith("data: "): 

176 data = line[6:].strip() 

177 import json 

178 

179 try: 

180 event = json.loads(data) 

181 event_type = event.get("type", "") 

182 if event_type == "content_block_delta": 

183 delta = event.get("delta", {}) 

184 yield { 

185 "delta": delta.get("text", ""), 

186 "type": delta.get("type", "text"), 

187 } 

188 elif event_type == "message_stop": 

189 yield {"delta": "", "finish_reason": "stop"} 

190 break 

191 elif event_type == "error": 

192 yield {"error": event.get("error", {}).get("message", "")} 

193 break 

194 except json.JSONDecodeError: 

195 continue 

196 

197 def sync_chat( 

198 self, 

199 messages: list[dict[str, str]], 

200 system: str | None = None, 

201 **kwargs, 

202 ) -> dict[str, Any]: 

203 """Synchronous chat completion.""" 

204 import asyncio 

205 

206 loop = asyncio.new_event_loop() 

207 try: 

208 return loop.run_until_complete(self.chat(messages, system, **kwargs)) 

209 finally: 

210 loop.close()