from __future__ import annotations import unittest from types import SimpleNamespace from unittest.mock import AsyncMock from micro_agents import Conversation, LlmMessage, Tool, ToolCallMessage, ToolCallResponseMessage, UserMessage from micro_agents.conversation import ReasoningMessage from app.llm import AnthropicMessagesLLM, ChatCompletionsLLM, OpenRouterChatLLM class ChatCompletionsLlmTests(unittest.IsolatedAsyncioTestCase): async def test_uses_chat_completions_with_system_prompt_and_thinking(self) -> None: response_message = SimpleNamespace( content="done", reasoning_content="reasoning", tool_calls=None, ) create = AsyncMock( return_value=SimpleNamespace( choices=[SimpleNamespace(message=response_message)], ) ) llm = ChatCompletionsLLM("https://glm.example/v4", "key", "glm-model") llm._ai = SimpleNamespace( # type: ignore[assignment] chat=SimpleNamespace(completions=SimpleNamespace(create=create)) ) result = await llm.ask( Conversation([UserMessage("hello")]), [], instructions="system prompt", thinking=True, ) request = create.await_args.kwargs self.assertEqual(request["model"], "glm-model") self.assertEqual( request["messages"], [ {"role": "system", "content": "system prompt"}, {"role": "user", "content": "hello"}, ], ) self.assertEqual(request["extra_body"], {"thinking": {"type": "enabled"}}) self.assertIsInstance(result[0], ReasoningMessage) self.assertIsInstance(result[1], LlmMessage) async def test_serializes_and_parses_tool_calls(self) -> None: model_call = SimpleNamespace( id="call-2", function=SimpleNamespace(name="lookup", arguments='{"query":"news"}'), ) create = AsyncMock( return_value=SimpleNamespace( choices=[ SimpleNamespace( message=SimpleNamespace( content="", reasoning_content=None, tool_calls=[model_call], ) ) ], ) ) llm = ChatCompletionsLLM("https://glm.example/v4", "key", "glm-model") llm._ai = SimpleNamespace( # type: ignore[assignment] chat=SimpleNamespace(completions=SimpleNamespace(create=create)) ) def lookup(query: str) -> str: """Look up a query.""" return query conversation = Conversation( [ UserMessage("find it"), ReasoningMessage("need a tool"), LlmMessage("checking"), ToolCallMessage("call-1", "lookup", {"query": "weather"}), ToolCallResponseMessage("call-1", "sunny"), ] ) result = await llm.ask(conversation, [Tool(lookup)], thinking=False) request = create.await_args.kwargs self.assertEqual(request["extra_body"], {"thinking": {"type": "disabled"}}) self.assertEqual(request["tools"][0]["type"], "function") self.assertEqual(request["tools"][0]["function"]["name"], "lookup") self.assertEqual(request["messages"][1]["role"], "assistant") self.assertEqual(request["messages"][1]["reasoning_content"], "need a tool") self.assertEqual(request["messages"][2]["role"], "tool") self.assertIsInstance(result[0], ToolCallMessage) self.assertEqual(result[0].tool_name(), "lookup") self.assertEqual(result[0].arguments(), {"query": "news"}) class OpenRouterChatLlmTests(unittest.IsolatedAsyncioTestCase): async def test_requests_reasoning_effort_and_echoes_details(self) -> None: details = [{"type": "reasoning.text", "text": "step", "signature": "sig"}] create = AsyncMock( return_value=SimpleNamespace( choices=[ SimpleNamespace( message=SimpleNamespace( content="done", reasoning="step", reasoning_details=details, tool_calls=None, model_extra={}, ) ) ] ) ) llm = OpenRouterChatLLM("https://openrouter.ai/api/v1", "key", "anthropic/claude-haiku-5.5") llm._ai = SimpleNamespace( # type: ignore[assignment] chat=SimpleNamespace(completions=SimpleNamespace(create=create)) ) conversation = Conversation( [ UserMessage("hello"), ReasoningMessage("step", provider_item={"reasoning_details": details}), LlmMessage("checking"), ToolCallMessage("call-1", "lookup", {"query": "weather"}), ToolCallResponseMessage("call-1", "sunny"), ] ) result = await llm.ask(conversation, [], thinking=True) request = create.await_args.kwargs self.assertEqual(request["model"], "anthropic/claude-haiku-5.5") self.assertEqual(request["extra_body"], {"reasoning": {"effort": "medium"}}) self.assertNotIn("thinking", request["extra_body"]) assistant = request["messages"][1] self.assertEqual(assistant["reasoning"], "step") self.assertEqual(assistant["reasoning_details"], details) self.assertNotIn("reasoning_content", assistant) self.assertIsInstance(result[0], ReasoningMessage) self.assertEqual(result[0].provider_item(), {"reasoning_details": details}) self.assertEqual(result[1].message(), "done") async def test_omits_reasoning_body_without_flag(self) -> None: create = AsyncMock( return_value=SimpleNamespace( choices=[ SimpleNamespace( message=SimpleNamespace( content="ok", reasoning=None, reasoning_content=None, reasoning_details=None, tool_calls=None, model_extra={}, ) ) ] ) ) llm = OpenRouterChatLLM("https://openrouter.ai/api/v1", "key", "anthropic/claude-haiku-5.5") llm._ai = SimpleNamespace( # type: ignore[assignment] chat=SimpleNamespace(completions=SimpleNamespace(create=create)) ) await llm.ask(Conversation([UserMessage("hello")]), [], thinking=False) self.assertNotIn("extra_body", create.await_args.kwargs) class AnthropicMessagesLlmTests(unittest.IsolatedAsyncioTestCase): async def test_sends_messages_api_request_and_parses_blocks(self) -> None: captured: list[dict[str, object]] = [] async def fake_post(payload: dict[str, object]) -> dict[str, object]: captured.append(payload) return { "content": [ { "type": "thinking", "thinking": "look it up", "signature": "sig", }, {"type": "text", "text": "done"}, { "type": "tool_use", "id": "toolu_1", "name": "lookup", "input": {"query": "news"}, }, ] } llm = AnthropicMessagesLLM("https://api.anthropic.com", "key", "claude-haiku-5-5") llm._post = fake_post # type: ignore[method-assign] def lookup(query: str) -> str: """Look up a query.""" return query thinking = { "type": "thinking", "thinking": "previous", "signature": "old-sig", } result = await llm.ask( Conversation( [ UserMessage("hello", media=[]), ReasoningMessage("previous", provider_item=thinking), LlmMessage("checking"), ToolCallMessage("toolu_0", "lookup", {"query": "weather"}), ToolCallResponseMessage("toolu_0", "sunny"), ] ), [Tool(lookup)], instructions="system prompt", thinking=True, ) payload = captured[0] self.assertEqual(llm._messages_url(), "https://api.anthropic.com/v1/messages") self.assertEqual(payload["model"], "claude-haiku-5-5") self.assertEqual(payload["system"], "system prompt") self.assertEqual(payload["thinking"], {"type": "adaptive", "display": "summarized"}) self.assertEqual(payload["tools"][0]["name"], "lookup") self.assertEqual(payload["tools"][0]["input_schema"]["type"], "object") self.assertEqual( payload["messages"], [ {"role": "user", "content": [{"type": "text", "text": "hello"}]}, { "role": "assistant", "content": [ thinking, {"type": "text", "text": "checking"}, { "type": "tool_use", "id": "toolu_0", "name": "lookup", "input": {"query": "weather"}, }, ], }, { "role": "user", "content": [ { "type": "tool_result", "tool_use_id": "toolu_0", "content": "sunny", } ], }, ], ) self.assertIsInstance(result[0], ReasoningMessage) self.assertEqual( result[0].provider_item(), {"type": "thinking", "thinking": "look it up", "signature": "sig"}, ) self.assertIsInstance(result[1], LlmMessage) self.assertEqual(result[1].message(), "done") self.assertIsInstance(result[2], ToolCallMessage) self.assertEqual(result[2].arguments(), {"query": "news"}) async def test_disables_thinking_without_flag(self) -> None: captured: list[dict[str, object]] = [] async def fake_post(payload: dict[str, object]) -> dict[str, object]: captured.append(payload) return {"content": [{"type": "text", "text": "ok"}]} llm = AnthropicMessagesLLM("https://api.anthropic.com/v1", "key", "claude-haiku-5-5") llm._post = fake_post # type: ignore[method-assign] await llm.ask(Conversation([UserMessage("hello")]), [], thinking=False) self.assertEqual(llm._messages_url(), "https://api.anthropic.com/v1/messages") self.assertEqual(captured[0]["thinking"], {"type": "disabled"}) self.assertNotIn("tools", captured[0]) if __name__ == "__main__": unittest.main()