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

1"""Base HTTP-based LLM Provider with shared OpenAI-compatible API logic.""" 

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 BaseHttpProvider(LLMProvider): 

17 """OpenAI-compatible HTTP API provider base class. 

18 

19 Subclasses override: provider_name, API_URL, _api_key_env, _default_model 

20 """ 

21 

22 API_URL: str = "" 

23 _api_key_env: str = "" 

24 _default_model: str = "" 

25 

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

27 super().__init__(model=model or self._default_model) 

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

29 

30 # ── Message conversion ── 

31 

32 @staticmethod 

33 def _messages_to_api(messages: list[Message]) -> list[dict]: 

34 api_msgs = [] 

35 for m in messages: 

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

37 if m.content: 

38 entry["content"] = m.content 

39 if m.tool_calls: 

40 entry["tool_calls"] = [ 

41 { 

42 "id": tc.id, 

43 "type": tc.type if hasattr(tc, "type") else "function", 

44 "function": { 

45 "name": tc.name, 

46 "arguments": tc.arguments, 

47 }, 

48 } 

49 for tc in m.tool_calls 

50 ] 

51 if m.tool_call_id: 

52 entry["tool_call_id"] = m.tool_call_id 

53 entry["content"] = m.content or "" 

54 api_msgs.append(entry) 

55 return api_msgs 

56 

57 @staticmethod 

58 def _tools_to_api(tools: list[Tool]) -> list[dict]: 

59 return [t.as_schema() for t in tools] 

60 

61 # ── API call ── 

62 

63 def _call_api(self, messages: list[Message], tools: Optional[list[Tool]] = None, 

64 temperature: float = 0.7) -> CompletionResult: 

65 body: dict = { 

66 "model": self.model, 

67 "messages": self._messages_to_api(messages), 

68 "stream": False, 

69 "temperature": temperature, 

70 } 

71 if tools: 

72 body["tools"] = self._tools_to_api(tools) 

73 

74 req = urllib.request.Request( 

75 self.API_URL, 

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

77 headers={ 

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

79 "Authorization": f"Bearer {self._api_key}", 

80 }, 

81 method="POST", 

82 ) 

83 

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

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

86 

87 choice = data["choices"][0] 

88 msg = choice["message"] 

89 

90 tool_calls = None 

91 if msg.get("tool_calls"): 

92 tool_calls = [ 

93 ToolCall( 

94 id=tc["id"], 

95 name=tc["function"]["name"], 

96 arguments=tc["function"]["arguments"], 

97 ) 

98 for tc in msg["tool_calls"] 

99 ] 

100 

101 return CompletionResult( 

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

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

104 choices=[CompletionChoice( 

105 index=0, 

106 message=Message( 

107 role=MessageRole.ASSISTANT, 

108 content=msg.get("content", ""), 

109 tool_calls=tool_calls, 

110 ), 

111 finish_reason=choice.get("finish_reason", "stop"), 

112 )], 

113 usage=CompletionUsage( 

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

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

116 total_tokens=data.get("usage", {}).get("total_tokens", 0), 

117 ), 

118 ) 

119 

120 # ── Interface ── 

121 

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

123 tools = kwargs.get("tools") 

124 temperature = kwargs.get("temperature", 0.7) 

125 return self._call_api(messages, tools=tools, temperature=temperature) 

126 

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

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