nanobot/hooks/server.py

@@ -0,0 +1,142 @@
+"""HTTP hooks server for external service integration."""
+
+import asyncio
+import json
+import uuid
+
+from aiohttp import web
+from loguru import logger
+
+from nanobot.bus.events import InboundMessage
+from nanobot.bus.queue import MessageBus
+from nanobot.config.schema import HooksConfig
+
+
+class HooksServer:
+    """
+    HTTP server exposing a /hooks endpoint.
+
+    External services POST JSON messages. The server publishes them
+    to the bus as InboundMessages and uses bus-level correlation
+    to return the agent's response synchronously.
+    """
+
+    def __init__(
+        self,
+        host: str,
+        port: int,
+        config: HooksConfig,
+        bus: MessageBus,
+    ):
+        self.host = host
+        self.port = port
+        self.config = config
+        self.bus = bus
+        self._app = web.Application()
+        self._app.router.add_post(self.config.path, self._handle_hook)
+        self._app.router.add_get("/health", self._handle_health)
+        self._runner: web.AppRunner | None = None
+
+    async def start(self) -> None:
+        """Start the HTTP server."""
+        if not self.config.has_tokens:
+            logger.warning("Hooks server has no tokens configured — endpoint disabled for security")
+            return
+
+        self._runner = web.AppRunner(self._app)
+        await self._runner.setup()
+        site = web.TCPSite(self._runner, self.host, self.port)
+        await site.start()
+        logger.info(f"Hooks server listening on {self.host}:{self.port}{self.config.path}")
+
+    async def stop(self) -> None:
+        """Stop the HTTP server."""
+        if self._runner:
+            await self._runner.cleanup()
+            self._runner = None
+
+    def _resolve_auth(self, request: web.Request) -> str | None:
+        """
+        Validate auth and return token name if valid, None otherwise.
+        Checks Authorization: Bearer <token> and X-Hook-Token headers.
+        """
+        auth = request.headers.get("Authorization", "")
+        if auth.startswith("Bearer "):
+            name = self.config.resolve_token(auth[7:])
+            if name:
+                return name
+
+        token = request.headers.get("X-Hook-Token", "")
+        if token:
+            name = self.config.resolve_token(token)
+            if name:
+                return name
+
+        return None
+
+    async def _handle_health(self, request: web.Request) -> web.Response:
+        """Health check endpoint — no auth required."""
+        return web.json_response({"status": "ok"})
+
+    async def _handle_hook(self, request: web.Request) -> web.Response:
+        """Handle incoming hook request."""
+        # Auth check — resolve token name
+        token_name = self._resolve_auth(request)
+        if not token_name:
+            return web.json_response({"error": "unauthorized"}, status=401)
+
+        # Parse body
+        try:
+            body = await request.json()
+        except (json.JSONDecodeError, Exception):
+            return web.json_response({"error": "invalid JSON body"}, status=400)
+
+        # Validate required fields
+        message = body.get("message")
+        if not message or not isinstance(message, str):
+            return web.json_response(
+                {"error": "missing or invalid 'message' field"}, status=400
+            )
+
+        # Optional fields
+        channel = body.get("channel", "hook")
+        chat_id = body.get("chat_id", token_name)
+        timeout = body.get("timeout", self.config.timeout_seconds)
+
+        # Create correlation
+        correlation_id = str(uuid.uuid4())
+
+        # Build InboundMessage
+        msg = InboundMessage(
+            channel=channel,
+            sender_id=f"hook:{token_name}",
+            chat_id=str(chat_id),
+            content=message,
+            metadata={
+                "correlation_id": correlation_id,
+                "hook_source": token_name,
+            },
+        )
+
+        # Fire-and-forget mode
+        if timeout == 0:
+            await self.bus.publish_inbound(msg)
+            return web.json_response({"ok": True}, status=202)
+
+        # Request-response mode
+        future = self.bus.register_correlation(correlation_id)
+        await self.bus.publish_inbound(msg)
+
+        try:
+            response = await asyncio.wait_for(future, timeout=timeout)
+            return web.json_response({"ok": True, "response": response})
+        except asyncio.TimeoutError:
+            self.bus.cancel_correlation(correlation_id)
+            return web.json_response(
+                {"ok": False, "error": f"agent did not respond within {timeout}s"},
+                status=504,
+            )
+        except Exception as e:
+            self.bus.cancel_correlation(correlation_id)
+            logger.error(f"Hook processing error: {e}")
+            return web.json_response({"error": "internal error"}, status=500)

tests/test_hooks_server.py

@@ -0,0 +1,176 @@
+"""Tests for the rewritten hooks server."""
+
+import asyncio
+import json
+import pytest
+from aiohttp import web
+from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop, TestClient, TestServer
+from nanobot.hooks.server import HooksServer
+from nanobot.bus.queue import MessageBus
+from nanobot.bus.events import InboundMessage, OutboundMessage
+from nanobot.config.schema import HooksConfig
+
+
+@pytest.fixture
+def bus():
+    return MessageBus()
+
+
+@pytest.fixture
+def config():
+    return HooksConfig(
+        enabled=True,
+        tokens={"test-hook": "test-secret-123"},
+        timeout_seconds=5,
+    )
+
+
+@pytest.fixture
+def server(bus, config):
+    return HooksServer(host="127.0.0.1", port=0, config=config, bus=bus)
+
+
+@pytest.mark.asyncio
+async def test_health_check(server):
+    client = TestClient(TestServer(server._app))
+    async with client:
+        resp = await client.get("/health")
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["status"] == "ok"
+
+
+@pytest.mark.asyncio
+async def test_unauthorized_without_token(server):
+    client = TestClient(TestServer(server._app))
+    async with client:
+        resp = await client.post("/hooks", json={"message": "test"})
+        assert resp.status == 401
+
+
+@pytest.mark.asyncio
+async def test_unauthorized_wrong_token(server):
+    client = TestClient(TestServer(server._app))
+    async with client:
+        resp = await client.post(
+            "/hooks",
+            json={"message": "test"},
+            headers={"Authorization": "Bearer wrong-token"},
+        )
+        assert resp.status == 401
+
+
+@pytest.mark.asyncio
+async def test_missing_message_field(server, bus):
+    client = TestClient(TestServer(server._app))
+    async with client:
+        resp = await client.post(
+            "/hooks",
+            json={"not_message": "test"},
+            headers={"Authorization": "Bearer test-secret-123"},
+        )
+        assert resp.status == 400
+
+
+@pytest.mark.asyncio
+async def test_hook_publishes_to_bus(server, bus):
+    """Hook should publish InboundMessage to bus and the message should contain hook prefix."""
+    client = TestClient(TestServer(server._app))
+    async with client:
+        # Send hook request in background (it will block waiting for correlation)
+        async def send_request():
+            return await client.post(
+                "/hooks",
+                json={"message": "hello from webhook"},
+                headers={"Authorization": "Bearer test-secret-123"},
+            )
+
+        task = asyncio.create_task(send_request())
+
+        # Consume the inbound message
+        msg = await asyncio.wait_for(bus.consume_inbound(), timeout=2.0)
+
+        assert msg.channel == "hook"
+        assert msg.chat_id == "test-hook"  # defaults to token name
+        assert msg.metadata.get("hook_source") == "test-hook"
+        assert msg.metadata.get("correlation_id") is not None
+
+        # Simulate agent response by resolving correlation
+        bus.resolve_correlation(OutboundMessage(
+            channel="hook",
+            chat_id="test-hook",
+            content="agent says hi",
+            metadata={"correlation_id": msg.metadata["correlation_id"]},
+        ))
+
+        resp = await asyncio.wait_for(task, timeout=2.0)
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["ok"] is True
+        assert data["response"] == "agent says hi"
+
+
+@pytest.mark.asyncio
+async def test_hook_with_custom_channel(server, bus):
+    """Hook targeting telegram should use telegram channel in InboundMessage."""
+    client = TestClient(TestServer(server._app))
+    async with client:
+        async def send_request():
+            return await client.post(
+                "/hooks",
+                json={"message": "notify user", "channel": "telegram", "chat_id": "239824268"},
+                headers={"Authorization": "Bearer test-secret-123"},
+            )
+
+        task = asyncio.create_task(send_request())
+
+        msg = await asyncio.wait_for(bus.consume_inbound(), timeout=2.0)
+        assert msg.channel == "telegram"
+        assert msg.chat_id == "239824268"
+        assert msg.session_key == "telegram:239824268"
+
+        bus.resolve_correlation(OutboundMessage(
+            channel="telegram",
+            chat_id="239824268",
+            content="done",
+            metadata={"correlation_id": msg.metadata["correlation_id"]},
+        ))
+
+        resp = await asyncio.wait_for(task, timeout=2.0)
+        assert resp.status == 200
+        data = await resp.json()
+        assert data["response"] == "done"
+
+
+@pytest.mark.asyncio
+async def test_hook_timeout_returns_504(bus):
+    """If agent doesn't respond in time, return 504."""
+    config = HooksConfig(enabled=True, tokens={"test-hook": "test-secret-123"}, timeout_seconds=1)
+    server = HooksServer(host="127.0.0.1", port=0, config=config, bus=bus)
+    client = TestClient(TestServer(server._app))
+    async with client:
+        resp = await client.post(
+            "/hooks",
+            json={"message": "slow request"},
+            headers={"Authorization": "Bearer test-secret-123"},
+        )
+        assert resp.status == 504
+
+
+@pytest.mark.asyncio
+async def test_hook_timeout_zero_returns_202(server, bus):
+    """timeout=0 should return 202 immediately without waiting."""
+    client = TestClient(TestServer(server._app))
+    async with client:
+        resp = await client.post(
+            "/hooks",
+            json={"message": "fire and forget", "timeout": 0},
+            headers={"Authorization": "Bearer test-secret-123"},
+        )
+        assert resp.status == 202
+        data = await resp.json()
+        assert data["ok"] is True
+
+        # Message should still be on the bus
+        msg = await asyncio.wait_for(bus.consume_inbound(), timeout=1.0)
+        assert msg.content == "fire and forget"