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 

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

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

23 

24 Subclasses override: provider_name, API_URL, _api_key_env, _default_model 

25 """ 

26 

27 API_URL: str = "" 

28 _api_key_env: str = "" 

29 _default_model: str = "" 

30 

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

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

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

34 

35 # ── Message conversion ── 

36 

37 @staticmethod 

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

39 api_msgs = [] 

40 for m in messages: 

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

42 if m.content: 

43 entry["content"] = m.content 

44 if m.tool_calls: 

45 entry["tool_calls"] = [ 

46 { 

47 "id": tc.id, 

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

49 "function": { 

50 "name": tc.name, 

51 "arguments": tc.arguments, 

52 }, 

53 } 

54 for tc in m.tool_calls 

55 ] 

56 if m.tool_call_id: 

57 entry["tool_call_id"] = m.tool_call_id 

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

59 api_msgs.append(entry) 

60 return api_msgs 

61 

62 @staticmethod 

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

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

65 

66 # ── API call ── 

67 

68 def _call_api( 

69 self, messages: list[Message], tools: list[Tool] | None = None, temperature: float = 0.7 

70 ) -> CompletionResult: 

71 body: dict = { 

72 "model": self.model, 

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

74 "stream": False, 

75 "temperature": temperature, 

76 } 

77 if tools: 

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

79 

80 req = urllib.request.Request( 

81 self.API_URL, 

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

83 headers={ 

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

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

86 }, 

87 method="POST", 

88 ) 

89 

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

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

92 

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

94 msg = choice["message"] 

95 

96 tool_calls = None 

97 if msg.get("tool_calls"): 

98 tool_calls = [ 

99 ToolCall( 

100 id=tc["id"], 

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

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

103 ) 

104 for tc in msg["tool_calls"] 

105 ] 

106 

107 return CompletionResult( 

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

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

110 choices=[ 

111 CompletionChoice( 

112 index=0, 

113 message=Message( 

114 role=MessageRole.ASSISTANT, 

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

116 tool_calls=tool_calls, 

117 ), 

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

119 ) 

120 ], 

121 usage=CompletionUsage( 

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

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

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

125 ), 

126 ) 

127 

128 # ── Interface ── 

129 

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

131 tools = kwargs.get("tools") 

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

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

134 

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

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