Build session-based agent tool loop

packages/micro-agents-tools/tests/test_tools.py

@@ -0,0 +1,100 @@
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from micro_agents import OpenAIBasedLLM, ToolCallMessage, ToolManager
+from micro_agents_tools import BraveSearchTools, FilesystemTools, PythonTools
+
+
+class FakeBraveSearchTools(BraveSearchTools):
+    def _request(self, url):
+        self.url = url
+        return {
+            "web": {
+                "results": [
+                    {"title": "Example", "url": "https://example.com", "description": "A result"},
+                    {"title": "Missing URL"},
+                ]
+            }
+        }
+
+
+class FilesystemToolsTests(unittest.IsolatedAsyncioTestCase):
+    async def test_brave_search_returns_compact_results(self):
+        tools = FakeBraveSearchTools(api_key="test-key")
+        response = await ToolManager(tools.tools()).execute(
+            ToolCallMessage("search", "web_search", {"query": "micro agents", "count": 3}),
+            {},
+        )
+        payload = json.loads(response.response.text())
+
+        self.assertIn("q=micro+agents", tools.url)
+        self.assertEqual(
+            [{"title": "Example", "url": "https://example.com", "description": "A result"}],
+            payload["results"],
+        )
+
+    async def test_filesystem_tools_use_the_context_base_directory(self):
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            root = Path(temporary_directory) / "test"
+            root.mkdir()
+            manager = ToolManager(FilesystemTools().tools())
+
+            write = await manager.execute(
+                ToolCallMessage("write", "write_file", {"path": "nested/hello.txt", "content": "hello"}),
+                {"base_dir": root},
+            )
+            read = await manager.execute(
+                ToolCallMessage("read", "read_file", {"path": "nested/hello.txt"}),
+                {"base_dir": root},
+            )
+
+            self.assertEqual("Wrote nested/hello.txt", write.response.text())
+            self.assertEqual("hello", read.response.text())
+
+    async def test_filesystem_tools_reject_path_escapes(self):
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            root = Path(temporary_directory) / "test"
+            root.mkdir()
+            response = await ToolManager(FilesystemTools().tools()).execute(
+                ToolCallMessage("write", "write_file", {"path": "../outside.txt", "content": "nope"}),
+                {"base_dir": root},
+            )
+
+            self.assertTrue(response.response.is_error())
+            self.assertEqual("tool_execution_failed", response.response.error_code())
+
+    async def test_python_tool_uses_the_session_base_directory(self):
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            base_dir = Path(temporary_directory)
+            code = "from pathlib import Path; Path('created.txt').write_text('ok'); print(Path.cwd())"
+            response = await ToolManager(PythonTools().tools()).execute(
+                ToolCallMessage("call-1", "execute_python", {"code": code}),
+                {"base_dir": base_dir},
+            )
+            payload = json.loads(response.response.text())
+
+            self.assertEqual(0, payload["exit_code"])
+            self.assertFalse(payload["timed_out"])
+            self.assertEqual("ok", (base_dir / "created.txt").read_text())
+
+    async def test_attach_files_returns_model_attachments(self):
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            base_dir = Path(temporary_directory)
+            (base_dir / "note.txt").write_text("hello", encoding="utf-8")
+            response = await ToolManager(FilesystemTools().tools()).execute(
+                ToolCallMessage("attach", "attach_files", {"paths": ["note.txt"]}),
+                {"base_dir": base_dir},
+            )
+
+            self.assertEqual(["note.txt"], [item.filename for item in response.response.attachments()])
+            serialized = OpenAIBasedLLM._serialize_tool_call_response_message(response.response)
+            output = serialized["output"]
+            assert isinstance(output, list)
+            self.assertEqual("input_text", output[0]["type"])
+            self.assertEqual("input_file", output[1]["type"])
+
+
+if __name__ == "__main__":
+    unittest.main()