diff --git a/CHANGELOG.md b/CHANGELOG.md index 28b437b53..28ab2a4b3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -50,6 +50,12 @@ `database_url` resolves inside `CLEVERAGENTS_HOME`, not the current working directory. Includes Robot Framework integration tests with a helper script exercising the same resolution path via subprocess. (#1034) +- Added FastAPI-based ASGI server endpoint served by uvicorn for the + CleverAgents server mode. Includes health check endpoint (`/health`), + A2A Agent Card discovery (`/.well-known/agent.json`), A2A JSON-RPC 2.0 + routing (`/a2a`), configurable host:port binding via Settings, graceful + shutdown on SIGTERM/SIGINT, and `agents server start` CLI command. + Behave BDD tests and Robot Framework integration tests included. (#862) - Added BuiltinAdapter class and MCP automatic resource slot creation. BuiltinAdapter wraps register_file_tools/register_git_tools/register_subplan_tool into a unified adapter interface. McpAdapter.infer_resource_slots() analyzes diff --git a/features/server_lifecycle.feature b/features/server_lifecycle.feature new file mode 100644 index 000000000..f459bca9d --- /dev/null +++ b/features/server_lifecycle.feature @@ -0,0 +1,122 @@ +@phase2 @a2a @server +Feature: ASGI Server Lifecycle + As a platform operator + I want to start, health-check, and shut down the CleverAgents ASGI server + So that the A2A JSON-RPC 2.0 endpoint is available for client communication + + # ----------------------------------------------------------------------- + # ASGI application creation + # ----------------------------------------------------------------------- + + Scenario: create_asgi_app returns a FastAPI application + When I create an ASGI application with default settings + Then the ASGI app should be a FastAPI instance + + Scenario: create_asgi_app accepts a custom facade + Given an A2aLocalFacade with no services + When I create an ASGI application with that facade + Then the ASGI app should be a FastAPI instance + + Scenario: create_asgi_app rejects invalid facade type + When I try to create an ASGI application with facade "not-a-facade" + Then a TypeError should be raised for invalid facade + + Scenario: create_asgi_app rejects invalid port + When I try to create an ASGI application with port 0 + Then a ValueError should be raised for invalid port + + Scenario: create_asgi_app rejects invalid host + When I try to create an ASGI application with host "" + Then a ValueError should be raised for invalid host + + # ----------------------------------------------------------------------- + # Health check endpoint + # ----------------------------------------------------------------------- + + Scenario: Health endpoint returns healthy status + Given a running ASGI test client + When I request GET /health + Then the response status code should be 200 + And the response body should contain "healthy" + + # ----------------------------------------------------------------------- + # Agent Card discovery endpoint + # ----------------------------------------------------------------------- + + Scenario: Agent Card endpoint returns valid agent card + Given a running ASGI test client + When I request GET /.well-known/agent.json + Then the response status code should be 200 + And the response body should contain "CleverAgents" + And the response body should contain "url" + + # ----------------------------------------------------------------------- + # A2A JSON-RPC endpoint + # ----------------------------------------------------------------------- + + Scenario: A2A endpoint dispatches health check operation + Given a running ASGI test client + When I POST a JSON-RPC request to /a2a with operation "_cleveragents/health/check" + Then the response status code should be 200 + And the A2A response status should be "ok" + + Scenario: A2A endpoint returns error for unknown operation + Given a running ASGI test client + When I POST a JSON-RPC request to /a2a with operation "nonexistent.operation" + Then the response status code should be 404 + + Scenario: A2A endpoint returns 400 for malformed request + Given a running ASGI test client + When I POST a malformed JSON body to /a2a + Then the response status code should be 400 + + # ----------------------------------------------------------------------- + # ServerLifecycle construction + # ----------------------------------------------------------------------- + + Scenario: ServerLifecycle accepts valid configuration + When I create a ServerLifecycle with host "127.0.0.1" and port 9000 + Then the lifecycle host should be "127.0.0.1" + And the lifecycle port should be 9000 + And the lifecycle should not be started + And the lifecycle should not be stopped + + Scenario: ServerLifecycle rejects empty host + When I try to create a ServerLifecycle with host "" + Then a ValueError should be raised for invalid host + + Scenario: ServerLifecycle rejects invalid port 0 + When I try to create a ServerLifecycle with port 0 + Then a ValueError should be raised for invalid port + + Scenario: ServerLifecycle rejects port above 65535 + When I try to create a ServerLifecycle with port 70000 + Then a ValueError should be raised for invalid port + + Scenario: ServerLifecycle rejects empty log_level + When I try to create a ServerLifecycle with empty log_level + Then a ValueError should be raised for invalid log_level + + # ----------------------------------------------------------------------- + # Server configuration from Settings + # ----------------------------------------------------------------------- + + Scenario: Settings provides default server host + Then the default server host from Settings should be "0.0.0.0" + + Scenario: Settings provides default server port + Then the default server port from Settings should be 8080 + + # ----------------------------------------------------------------------- + # Graceful shutdown + # ----------------------------------------------------------------------- + + Scenario: ServerLifecycle request_shutdown sets flag + Given a ServerLifecycle with a mock uvicorn server + When I call request_shutdown on the lifecycle + Then the mock server should_exit flag should be true + + Scenario: ServerLifecycle cannot be started twice + Given a ServerLifecycle that has already been started + When I try to start the lifecycle again + Then a RuntimeError should be raised for double start diff --git a/features/steps/server_lifecycle_steps.py b/features/steps/server_lifecycle_steps.py new file mode 100644 index 000000000..35cff604b --- /dev/null +++ b/features/steps/server_lifecycle_steps.py @@ -0,0 +1,274 @@ +"""Step definitions for ``server_lifecycle.feature``. + +Tests the ASGI application factory, health/agent-card endpoints, +A2A JSON-RPC dispatch, ServerLifecycle construction and shutdown, +and Settings-based server configuration. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock + +from behave import given, then, when # type: ignore[import-untyped] + +from cleveragents.a2a.facade import A2aLocalFacade +from cleveragents.config.settings import Settings +from cleveragents.infrastructure.server.asgi_app import create_asgi_app +from cleveragents.infrastructure.server.server_lifecycle import ServerLifecycle + +# --------------------------------------------------------------------------- +# ASGI application creation +# --------------------------------------------------------------------------- + + +@when("I create an ASGI application with default settings") +def step_create_asgi_default(context: Any) -> None: + context.asgi_app = create_asgi_app() + + +@given("an A2aLocalFacade with no services") +def step_given_facade(context: Any) -> None: + context.facade = A2aLocalFacade() + + +@when("I create an ASGI application with that facade") +def step_create_asgi_with_facade(context: Any) -> None: + context.asgi_app = create_asgi_app(facade=context.facade) + + +@then("the ASGI app should be a FastAPI instance") +def step_check_fastapi_instance(context: Any) -> None: + from fastapi import FastAPI + + assert isinstance(context.asgi_app, FastAPI), ( + f"Expected FastAPI, got {type(context.asgi_app)}" + ) + + +@when('I try to create an ASGI application with facade "{value}"') +def step_create_asgi_bad_facade(context: Any, value: str) -> None: + context.caught_exception = None + try: + create_asgi_app(facade=value) # type: ignore[arg-type] + except TypeError as exc: + context.caught_exception = exc + + +@then("a TypeError should be raised for invalid facade") +def step_check_type_error_facade(context: Any) -> None: + assert context.caught_exception is not None, "Expected TypeError" + assert isinstance(context.caught_exception, TypeError) + + +@when("I try to create an ASGI application with port {port:d}") +def step_create_asgi_bad_port(context: Any, port: int) -> None: + context.caught_exception = None + try: + create_asgi_app(port=port) + except ValueError as exc: + context.caught_exception = exc + + +@then("a ValueError should be raised for invalid port") +def step_check_value_error_port(context: Any) -> None: + assert context.caught_exception is not None, "Expected ValueError" + assert isinstance(context.caught_exception, ValueError) + + +@when('I try to create an ASGI application with host "{host}"') +def step_create_asgi_bad_host(context: Any, host: str) -> None: + context.caught_exception = None + try: + create_asgi_app(host=host) + except ValueError as exc: + context.caught_exception = exc + + +@then("a ValueError should be raised for invalid host") +def step_check_value_error_host(context: Any) -> None: + assert context.caught_exception is not None, "Expected ValueError" + assert isinstance(context.caught_exception, ValueError) + + +# --------------------------------------------------------------------------- +# Test client helpers +# --------------------------------------------------------------------------- + + +@given("a running ASGI test client") +def step_given_test_client(context: Any) -> None: + from fastapi.testclient import TestClient + + app = create_asgi_app() + context.test_client = TestClient(app) + + +@when("I request GET /health") +def step_get_health(context: Any) -> None: + context.response = context.test_client.get("/health") + + +@when("I request GET /.well-known/agent.json") +def step_get_agent_card(context: Any) -> None: + context.response = context.test_client.get("/.well-known/agent.json") + + +@when('I POST a JSON-RPC request to /a2a with operation "{operation}"') +def step_post_a2a(context: Any, operation: str) -> None: + payload = {"operation": operation, "params": {}} + context.response = context.test_client.post("/a2a", json=payload) + + +@when("I POST a malformed JSON body to /a2a") +def step_post_a2a_malformed(context: Any) -> None: + # Send a body that cannot be parsed into A2aRequest (missing operation) + context.response = context.test_client.post("/a2a", json={"bad": "data"}) + + +@then("the response status code should be {code:d}") +def step_check_status_code(context: Any, code: int) -> None: + assert context.response.status_code == code, ( + f"Expected {code}, got {context.response.status_code}: {context.response.text}" + ) + + +@then('the response body should contain "{text}"') +def step_check_response_body(context: Any, text: str) -> None: + body = context.response.text + assert text in body, f"Expected '{text}' in response body: {body}" + + +@then('the A2A response status should be "{expected_status}"') +def step_check_a2a_status(context: Any, expected_status: str) -> None: + data = context.response.json() + assert data.get("status") == expected_status, ( + f"Expected A2A status '{expected_status}', got {data}" + ) + + +# --------------------------------------------------------------------------- +# ServerLifecycle construction +# --------------------------------------------------------------------------- + + +@when('I create a ServerLifecycle with host "{host}" and port {port:d}') +def step_create_lifecycle(context: Any, host: str, port: int) -> None: + context.lifecycle = ServerLifecycle(host=host, port=port) + + +@then('the lifecycle host should be "{expected}"') +def step_check_lifecycle_host(context: Any, expected: str) -> None: + assert context.lifecycle.host == expected + + +@then("the lifecycle port should be {expected:d}") +def step_check_lifecycle_port(context: Any, expected: int) -> None: + assert context.lifecycle.port == expected + + +@then("the lifecycle should not be started") +def step_check_not_started(context: Any) -> None: + assert not context.lifecycle.is_started + + +@then("the lifecycle should not be stopped") +def step_check_not_stopped(context: Any) -> None: + assert not context.lifecycle.is_stopped + + +@when('I try to create a ServerLifecycle with host "{host}"') +def step_create_lifecycle_bad_host(context: Any, host: str) -> None: + context.caught_exception = None + try: + ServerLifecycle(host=host) + except ValueError as exc: + context.caught_exception = exc + + +@when("I try to create a ServerLifecycle with port {port:d}") +def step_create_lifecycle_bad_port(context: Any, port: int) -> None: + context.caught_exception = None + try: + ServerLifecycle(port=port) + except ValueError as exc: + context.caught_exception = exc + + +@when("I try to create a ServerLifecycle with empty log_level") +def step_create_lifecycle_bad_log_level(context: Any) -> None: + context.caught_exception = None + try: + ServerLifecycle(log_level="") + except ValueError as exc: + context.caught_exception = exc + + +@then("a ValueError should be raised for invalid log_level") +def step_check_value_error_log_level(context: Any) -> None: + assert context.caught_exception is not None, "Expected ValueError" + assert isinstance(context.caught_exception, ValueError) + + +# --------------------------------------------------------------------------- +# Settings +# --------------------------------------------------------------------------- + + +@then('the default server host from Settings should be "{expected}"') +def step_check_settings_host(context: Any, expected: str) -> None: + settings = Settings() + assert settings.server_host == expected + + +@then("the default server port from Settings should be {expected:d}") +def step_check_settings_port(context: Any, expected: int) -> None: + settings = Settings() + assert settings.server_port == expected + + +# --------------------------------------------------------------------------- +# Graceful shutdown +# --------------------------------------------------------------------------- + + +@given("a ServerLifecycle with a mock uvicorn server") +def step_lifecycle_with_mock_server(context: Any) -> None: + lifecycle = ServerLifecycle(host="127.0.0.1", port=9999) + mock_server = MagicMock() + mock_server.should_exit = False + lifecycle._server = mock_server + context.lifecycle = lifecycle + context.mock_server = mock_server + + +@when("I call request_shutdown on the lifecycle") +def step_call_request_shutdown(context: Any) -> None: + context.lifecycle.request_shutdown() + + +@then("the mock server should_exit flag should be true") +def step_check_should_exit(context: Any) -> None: + assert context.mock_server.should_exit is True + + +@given("a ServerLifecycle that has already been started") +def step_lifecycle_already_started(context: Any) -> None: + lifecycle = ServerLifecycle(host="127.0.0.1", port=9998) + lifecycle._started = True + context.lifecycle = lifecycle + + +@when("I try to start the lifecycle again") +def step_try_double_start(context: Any) -> None: + context.caught_exception = None + try: + context.lifecycle.start() + except RuntimeError as exc: + context.caught_exception = exc + + +@then("a RuntimeError should be raised for double start") +def step_check_runtime_error(context: Any) -> None: + assert context.caught_exception is not None, "Expected RuntimeError" + assert isinstance(context.caught_exception, RuntimeError) diff --git a/pyproject.toml b/pyproject.toml index ad642ab01..68732c251 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,6 +26,7 @@ dependencies = [ # CLI Framework (ADR-009) "typer>=0.9.0", "uvicorn>=0.30.1", + "fastapi>=0.115.0", "watchdog>=4.0.0", "faiss-cpu>=1.7.4", # Vector store backend "rx>=3.2.0", # Reactive streams for routing diff --git a/robot/helper_server_lifecycle.py b/robot/helper_server_lifecycle.py new file mode 100644 index 000000000..82097fcfa --- /dev/null +++ b/robot/helper_server_lifecycle.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python3 +"""Integration test helper for server lifecycle tests. + +Each sub-command exercises a real import / instantiation path and prints +a unique sentinel on success. Robot tests parse the output for that +sentinel. +""" + +from __future__ import annotations + +import sys + + +def _test_create_app() -> None: + """Verify ``create_asgi_app`` returns a FastAPI instance.""" + from cleveragents.infrastructure.server.asgi_app import create_asgi_app + + app = create_asgi_app() + assert app is not None + print("create-app-ok") + + +def _test_health_endpoint() -> None: + """Exercise the /health endpoint via FastAPI TestClient.""" + from fastapi.testclient import TestClient + + from cleveragents.infrastructure.server.asgi_app import create_asgi_app + + app = create_asgi_app() + client = TestClient(app) + resp = client.get("/health") + assert resp.status_code == 200 + data = resp.json() + assert data["status"] == "healthy" + print("health-endpoint-ok") + + +def _test_agent_card() -> None: + """Exercise the /.well-known/agent.json endpoint.""" + from fastapi.testclient import TestClient + + from cleveragents.infrastructure.server.asgi_app import create_asgi_app + + app = create_asgi_app() + client = TestClient(app) + resp = client.get("/.well-known/agent.json") + assert resp.status_code == 200 + data = resp.json() + assert data["name"] == "CleverAgents" + assert "url" in data + print("agent-card-ok") + + +def _test_a2a_dispatch() -> None: + """Exercise the /a2a POST endpoint with a health check operation.""" + from fastapi.testclient import TestClient + + from cleveragents.infrastructure.server.asgi_app import create_asgi_app + + app = create_asgi_app() + client = TestClient(app) + payload = {"operation": "_cleveragents/health/check", "params": {}} + resp = client.post("/a2a", json=payload) + assert resp.status_code == 200 + data = resp.json() + assert data["status"] == "ok" + print("a2a-dispatch-ok") + + +def _test_lifecycle_construction() -> None: + """Verify ServerLifecycle can be constructed and inspected.""" + from cleveragents.infrastructure.server.server_lifecycle import ServerLifecycle + + lc = ServerLifecycle(host="127.0.0.1", port=9123) + assert lc.host == "127.0.0.1" + assert lc.port == 9123 + assert not lc.is_started + assert not lc.is_stopped + print("lifecycle-construction-ok") + + +def _test_lifecycle_shutdown() -> None: + """Verify request_shutdown sets the should_exit flag on a mock server.""" + from unittest.mock import MagicMock + + from cleveragents.infrastructure.server.server_lifecycle import ServerLifecycle + + lc = ServerLifecycle(host="127.0.0.1", port=9124) + mock_server = MagicMock() + mock_server.should_exit = False + lc._server = mock_server + lc.request_shutdown() + assert mock_server.should_exit is True + print("lifecycle-shutdown-ok") + + +_COMMANDS: dict[str, object] = { + "create-app": _test_create_app, + "health-endpoint": _test_health_endpoint, + "agent-card": _test_agent_card, + "a2a-dispatch": _test_a2a_dispatch, + "lifecycle-construction": _test_lifecycle_construction, + "lifecycle-shutdown": _test_lifecycle_shutdown, +} + + +def main() -> None: + if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS: + print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr) + sys.exit(2) + + handler = _COMMANDS[sys.argv[1]] + handler() # type: ignore[operator] + + +if __name__ == "__main__": + main() diff --git a/robot/server_lifecycle.robot b/robot/server_lifecycle.robot new file mode 100644 index 000000000..9fa2edbe8 --- /dev/null +++ b/robot/server_lifecycle.robot @@ -0,0 +1,61 @@ +*** Settings *** +Documentation Integration tests for ASGI server lifecycle +... +... Tests verify that the ASGI application, health endpoint, +... agent card, A2A dispatch, and ServerLifecycle construction +... all work end-to-end through real imports and instantiation. +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER} ${CURDIR}/helper_server_lifecycle.py + +*** Test Cases *** +Create ASGI Application + [Documentation] create_asgi_app returns a FastAPI instance + ${result}= Run Process ${PYTHON} ${HELPER} create-app cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} create-app-ok + +Health Endpoint Returns Healthy + [Documentation] GET /health returns 200 with status healthy + ${result}= Run Process ${PYTHON} ${HELPER} health-endpoint cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} health-endpoint-ok + +Agent Card Discovery Endpoint + [Documentation] GET /.well-known/agent.json returns valid agent card + ${result}= Run Process ${PYTHON} ${HELPER} agent-card cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} agent-card-ok + +A2A JSON-RPC Dispatch + [Documentation] POST /a2a dispatches health check operation via facade + ${result}= Run Process ${PYTHON} ${HELPER} a2a-dispatch cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} a2a-dispatch-ok + +ServerLifecycle Construction + [Documentation] ServerLifecycle construction sets host, port, and initial state + ${result}= Run Process ${PYTHON} ${HELPER} lifecycle-construction cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} lifecycle-construction-ok + +ServerLifecycle Graceful Shutdown + [Documentation] request_shutdown sets the should_exit flag + ${result}= Run Process ${PYTHON} ${HELPER} lifecycle-shutdown cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} lifecycle-shutdown-ok diff --git a/src/cleveragents/cli/commands/server.py b/src/cleveragents/cli/commands/server.py index 7d66dcfce..bd4b6c370 100644 --- a/src/cleveragents/cli/commands/server.py +++ b/src/cleveragents/cli/commands/server.py @@ -1,8 +1,8 @@ -"""Server connection CLI commands. +"""Server CLI commands — connection management and ASGI server start. -Provides the ``agents server connect`` stub command and server-mode -status helpers. All actual server communication is stubbed out — -commands persist configuration and print explicit warnings. +Provides the ``agents server connect`` stub command, the +``agents server start`` command that launches the uvicorn-backed +ASGI server, and server-mode status helpers. """ from __future__ import annotations @@ -20,7 +20,7 @@ from cleveragents.application.container import get_container from cleveragents.application.services.config_service import ConfigService from cleveragents.cli.formatting import OutputFormat, format_output -app: Any = typer.Typer(help="Server connection management (stub).") +app: Any = typer.Typer(help="Server management — start, connect, and status.") console: Console = Console() # --------------------------------------------------------------------------- @@ -213,9 +213,47 @@ def server_status( ) +@app.command("start") +def server_start( + host: Annotated[ + str | None, + typer.Option("--host", "-H", help="Bind address (default: from settings)"), + ] = None, + port: Annotated[ + int | None, + typer.Option("--port", "-p", help="Bind port (default: from settings)"), + ] = None, + log_level: Annotated[ + str | None, + typer.Option("--log-level", help="Uvicorn log level"), + ] = None, +) -> None: + """Start the CleverAgents ASGI server. + + Launches a uvicorn server hosting the A2A JSON-RPC 2.0 endpoint. + The server binds to host:port from Settings or CLI overrides and + handles graceful shutdown on SIGTERM/SIGINT. + + Examples:: + + agents server start + agents server start --host 127.0.0.1 --port 9000 + agents server start --log-level debug + """ + from cleveragents.infrastructure.server import run_server + + console.print("[bold]Starting CleverAgents ASGI server...[/bold]") + run_server( + host=host, + port=port, + log_level=log_level, + ) + + __all__ = [ "app", "resolve_server_mode", "server_connect", + "server_start", "server_status", ] diff --git a/src/cleveragents/cli/main.py b/src/cleveragents/cli/main.py index 7afcb56c1..2a2eccf94 100644 --- a/src/cleveragents/cli/main.py +++ b/src/cleveragents/cli/main.py @@ -206,7 +206,7 @@ def _register_subcommands() -> None: app.add_typer( server_app, name="server", - help="Server connection management (stub)", + help="Server management — start, connect, and status", ) app.add_typer( repo.app, diff --git a/src/cleveragents/infrastructure/server/__init__.py b/src/cleveragents/infrastructure/server/__init__.py new file mode 100644 index 000000000..9f38f8081 --- /dev/null +++ b/src/cleveragents/infrastructure/server/__init__.py @@ -0,0 +1,25 @@ +"""ASGI server infrastructure for CleverAgents. + +Provides the FastAPI-based ASGI application, A2A JSON-RPC 2.0 endpoint +routing, and uvicorn server lifecycle management. The server's sole +client-facing interface is the A2A JSON-RPC 2.0 endpoint — there is no +REST API (ADR-048). + +Public surface: + +- :func:`create_asgi_app` — build a configured FastAPI application +- :func:`run_server` — launch uvicorn with graceful shutdown +- :class:`ServerLifecycle` — start/stop lifecycle manager +""" + +from cleveragents.infrastructure.server.asgi_app import create_asgi_app +from cleveragents.infrastructure.server.server_lifecycle import ( + ServerLifecycle, + run_server, +) + +__all__ = [ + "ServerLifecycle", + "create_asgi_app", + "run_server", +] diff --git a/src/cleveragents/infrastructure/server/asgi_app.py b/src/cleveragents/infrastructure/server/asgi_app.py new file mode 100644 index 000000000..cbca9debe --- /dev/null +++ b/src/cleveragents/infrastructure/server/asgi_app.py @@ -0,0 +1,171 @@ +"""FastAPI-based ASGI application for the CleverAgents server. + +The server's sole client-facing interface is an A2A JSON-RPC 2.0 +endpoint (ADR-048). This module builds the FastAPI application, +wires the health-check / Agent Card discovery endpoint, and exposes +the A2A JSON-RPC dispatch route. + +All A2A operations — both standard (``message/send``, +``message/stream``) and ``_cleveragents/`` extension methods — flow +through the single ``/a2a`` POST endpoint. The Agent Card is served +at ``/.well-known/agent.json`` per the A2A specification. +""" + +from __future__ import annotations + +from typing import Any + +import structlog +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse + +from cleveragents.a2a.errors import A2aOperationNotFoundError +from cleveragents.a2a.facade import A2aLocalFacade +from cleveragents.a2a.models import A2aRequest, A2aVersion + +logger: structlog.stdlib.BoundLogger = structlog.get_logger(__name__) + +# ------------------------------------------------------------------ +# Agent Card (served at /.well-known/agent.json per A2A spec) +# ------------------------------------------------------------------ + +_AGENT_CARD: dict[str, Any] = { + "name": "CleverAgents", + "description": "AI-powered development assistant (actor-first)", + "url": "", + "version": A2aVersion.CURRENT, + "capabilities": { + "streaming": False, + "pushNotifications": False, + }, + "skills": [], + "defaultInputModes": ["text"], + "defaultOutputModes": ["text"], +} + + +def _build_agent_card(host: str, port: int) -> dict[str, Any]: + """Return the Agent Card with the server URL populated.""" + card = dict(_AGENT_CARD) + card["url"] = f"http://{host}:{port}/a2a" + return card + + +# ------------------------------------------------------------------ +# Application factory +# ------------------------------------------------------------------ + + +def create_asgi_app( + *, + facade: A2aLocalFacade | None = None, + host: str = "0.0.0.0", + port: int = 8080, +) -> FastAPI: + """Create and return a configured FastAPI ASGI application. + + Parameters + ---------- + facade: + The :class:`A2aLocalFacade` used to dispatch A2A operations. + When *None* a default facade with no services is created. + host: + Bind address used to populate the Agent Card URL. + port: + Bind port used to populate the Agent Card URL. + + Returns + ------- + FastAPI + A fully wired ASGI application ready for ``uvicorn.run()``. + """ + if facade is not None and not isinstance(facade, A2aLocalFacade): + raise TypeError("facade must be an A2aLocalFacade instance or None") + if not isinstance(host, str) or not host: + raise ValueError("host must be a non-empty string") + if not isinstance(port, int) or port < 1 or port > 65535: + raise ValueError("port must be an integer between 1 and 65535") + + resolved_facade = facade if facade is not None else A2aLocalFacade() + agent_card = _build_agent_card(host, port) + + application = FastAPI( + title="CleverAgents A2A Server", + description="A2A JSON-RPC 2.0 endpoint for CleverAgents", + version=A2aVersion.CURRENT, + docs_url=None, + redoc_url=None, + openapi_url=None, + ) + + # --- Health / Agent Card endpoint --- + + @application.get("/.well-known/agent.json") + async def agent_card_endpoint() -> JSONResponse: + """Serve the A2A Agent Card for capability discovery.""" + return JSONResponse(content=agent_card) + + @application.get("/health") + async def health_endpoint() -> JSONResponse: + """Simple liveness probe.""" + return JSONResponse(content={"status": "healthy"}) + + # --- A2A JSON-RPC 2.0 endpoint --- + + @application.post("/a2a") + async def a2a_endpoint(request: Request) -> JSONResponse: + """Handle an inbound A2A JSON-RPC 2.0 request. + + Accepts a JSON body conforming to the :class:`A2aRequest` + schema, dispatches it through the facade, and returns the + :class:`A2aResponse` envelope. + """ + body: dict[str, Any] = await request.json() + + try: + a2a_request = A2aRequest(**body) + except Exception as exc: + logger.warning("a2a.server.bad_request", error=str(exc)) + return JSONResponse( + status_code=400, + content={ + "jsonrpc": "2.0", + "error": { + "code": -32600, + "message": f"Invalid A2A request: {exc}", + }, + }, + ) + + try: + response = resolved_facade.dispatch(a2a_request) + return JSONResponse(content=response.model_dump()) + except A2aOperationNotFoundError as exc: + logger.warning( + "a2a.server.method_not_found", + operation=a2a_request.operation, + ) + return JSONResponse( + status_code=404, + content={ + "jsonrpc": "2.0", + "error": { + "code": -32601, + "message": str(exc), + }, + }, + ) + + logger.info( + "a2a.server.app_created", + host=host, + port=port, + operations=len(resolved_facade.list_operations()), + ) + + return application + + +__all__ = [ + "create_asgi_app", +] diff --git a/src/cleveragents/infrastructure/server/server_lifecycle.py b/src/cleveragents/infrastructure/server/server_lifecycle.py new file mode 100644 index 000000000..5f848a3f7 --- /dev/null +++ b/src/cleveragents/infrastructure/server/server_lifecycle.py @@ -0,0 +1,208 @@ +"""Uvicorn server lifecycle management with graceful shutdown. + +Provides :class:`ServerLifecycle` which wraps ``uvicorn.Server`` with +SIGTERM / SIGINT handling so that the server shuts down cleanly when +the process is terminated or the user presses Ctrl-C. + +The public :func:`run_server` convenience function creates the ASGI +application, configures uvicorn, and blocks until shutdown. +""" + +from __future__ import annotations + +import signal +import threading +from types import FrameType + +import structlog +import uvicorn + +from cleveragents.a2a.facade import A2aLocalFacade +from cleveragents.config.settings import Settings +from cleveragents.infrastructure.server.asgi_app import create_asgi_app + +logger: structlog.stdlib.BoundLogger = structlog.get_logger(__name__) + + +class ServerLifecycle: + """Manage the lifecycle of the uvicorn ASGI server. + + Parameters + ---------- + host: + Network address to bind to. + port: + TCP port to listen on. + facade: + The A2A facade used for request dispatch. *None* creates a + default facade with no wired services. + log_level: + Uvicorn log level (``"info"``, ``"warning"``, etc.). + """ + + def __init__( + self, + *, + host: str = "0.0.0.0", + port: int = 8080, + facade: A2aLocalFacade | None = None, + log_level: str = "info", + ) -> None: + if not isinstance(host, str) or not host: + raise ValueError("host must be a non-empty string") + if not isinstance(port, int) or port < 1 or port > 65535: + raise ValueError("port must be an integer between 1 and 65535") + if not isinstance(log_level, str) or not log_level: + raise ValueError("log_level must be a non-empty string") + + self._host: str = host + self._port: int = port + self._facade: A2aLocalFacade = ( + facade if facade is not None else A2aLocalFacade() + ) + self._log_level: str = log_level.lower() + self._server: uvicorn.Server | None = None + self._started: bool = False + self._stopped: bool = False + + # ------------------------------------------------------------------ + # Properties + # ------------------------------------------------------------------ + + @property + def host(self) -> str: + """Bound host address.""" + return self._host + + @property + def port(self) -> int: + """Bound port number.""" + return self._port + + @property + def is_started(self) -> bool: + """Whether :meth:`start` has been called.""" + return self._started + + @property + def is_stopped(self) -> bool: + """Whether the server has stopped.""" + return self._stopped + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + def start(self) -> None: + """Build the ASGI app and run uvicorn (blocking). + + Installs SIGTERM/SIGINT handlers for graceful shutdown. This + method blocks until the server exits. + """ + if self._started: + raise RuntimeError("Server has already been started") + + self._started = True + app = create_asgi_app( + facade=self._facade, + host=self._host, + port=self._port, + ) + + config = uvicorn.Config( + app=app, + host=self._host, + port=self._port, + log_level=self._log_level, + ) + server = uvicorn.Server(config) + self._server = server + + logger.info( + "a2a.server.starting", + host=self._host, + port=self._port, + ) + + # Install signal handlers for graceful shutdown. + self._install_signal_handlers() + + server.run() + self._stopped = True + + logger.info("a2a.server.stopped") + + def request_shutdown(self) -> None: + """Request a graceful shutdown of the running server. + + Safe to call from signal handlers or other threads. + """ + if self._server is not None: + self._server.should_exit = True + logger.info("a2a.server.shutdown_requested") + + # ------------------------------------------------------------------ + # Signal handling + # ------------------------------------------------------------------ + + def _install_signal_handlers(self) -> None: + """Install SIGTERM and SIGINT handlers when running on main thread.""" + if threading.current_thread() is not threading.main_thread(): + return + + def _handle_signal(signum: int, frame: FrameType | None) -> None: + sig_name = signal.Signals(signum).name + logger.info("a2a.server.signal_received", signal=sig_name) + self.request_shutdown() + + signal.signal(signal.SIGTERM, _handle_signal) + signal.signal(signal.SIGINT, _handle_signal) + + +# ------------------------------------------------------------------ +# Convenience runner +# ------------------------------------------------------------------ + + +def run_server( + *, + host: str | None = None, + port: int | None = None, + facade: A2aLocalFacade | None = None, + log_level: str | None = None, +) -> None: + """Create and run the ASGI server (blocking). + + Settings are resolved from :class:`Settings` when parameters are + *None*. + + Parameters + ---------- + host: + Override for ``Settings.server_host``. + port: + Override for ``Settings.server_port``. + facade: + A2A facade for operation dispatch. + log_level: + Override for ``Settings.log_level``. + """ + settings = Settings.get_settings() + + resolved_host = host if host is not None else settings.server_host + resolved_port = port if port is not None else settings.server_port + resolved_log_level = log_level if log_level is not None else settings.log_level + + lifecycle = ServerLifecycle( + host=resolved_host, + port=resolved_port, + facade=facade, + log_level=resolved_log_level, + ) + lifecycle.start() + + +__all__ = [ + "ServerLifecycle", + "run_server", +] diff --git a/vulture_whitelist.py b/vulture_whitelist.py index e47a0c35d..886599579 100644 --- a/vulture_whitelist.py +++ b/vulture_whitelist.py @@ -1190,3 +1190,9 @@ detect_directory_languages # noqa: B018, F821 get_servers_for_language # noqa: B018, F821 restart_server # noqa: B018, F821 _make_runtime_handler # noqa: B018, F821 +# ASGI server infrastructure — FastAPI route handlers and ServerLifecycle public API +agent_card_endpoint # noqa: B018, F821 +health_endpoint # noqa: B018, F821 +a2a_endpoint # noqa: B018, F821 +server_start # noqa: B018, F821 +run_server # noqa: B018, F821