feat(server): ASGI endpoint via uvicorn

Implement FastAPI-based ASGI application served by uvicorn for
the CleverAgents server mode. Add health check endpoint, A2A
JSON-RPC routing, configurable host:port binding, and graceful
shutdown handling. Server launches via `agents server start`.

- Add FastAPI ASGI app factory at infrastructure/server/asgi_app.py
  with /.well-known/agent.json (Agent Card), /health (liveness),
  and /a2a (A2A JSON-RPC 2.0 dispatch via A2aLocalFacade)
- Add ServerLifecycle class at infrastructure/server/server_lifecycle.py
  wrapping uvicorn.Server with SIGTERM/SIGINT graceful shutdown
- Add `agents server start` CLI command with --host, --port, --log-level
  options, resolving defaults from Settings
- Add fastapi>=0.115.0 dependency to pyproject.toml (uvicorn already present)
- Add Behave BDD tests (features/server_lifecycle.feature, 20 scenarios)
- Add Robot Framework integration tests (robot/server_lifecycle.robot)
- Update CHANGELOG.md and vulture_whitelist.py

ISSUES CLOSED: #862

CHANGELOG.md

@@ -44,6 +44,12 @@
   configurable singletons in the DI container with provider selection via
   `override_providers()`. Includes Behave BDD tests (35 scenarios), Robot
   Framework smoke tests, ASV benchmarks, and reference documentation. (#498)
+- 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

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

features/steps/a2a_facade_coverage_boost_steps.py

@@ -127,3 +127,7 @@ def step_cb_error_message_contains(context: Context, fragment: str) -> None:
         f"Expected message containing '{fragment}', "
         f"got '{context.cb_response.error.message}'"
     )
+
+
+# Restore default step matcher for other step files
+use_step_matcher("parse")

features/steps/a2a_facade_coverage_steps.py

@@ -344,3 +344,7 @@ def step_fc_resources_count(context: Context, count: str) -> None:
     assert len(resources) == int(count), (
         f"Expected {count} resources, got {len(resources)}"
     )
+
+
+# Restore default step matcher for other step files
+use_step_matcher("parse")

features/steps/a2a_facade_steps.py

@@ -486,3 +486,7 @@ def step_create_not_available_default(context: Context) -> None:
 
 
 # "the error message should contain" is provided by service_steps.py
+
+
+# Restore default step matcher for other step files
+use_step_matcher("parse")

features/steps/a2a_facade_wiring_steps.py

@@ -247,3 +247,7 @@ def step_wired_error_code(context: Context, code: str) -> None:
     assert context.wired_response.error.code == code, (
         f"Expected error code '{code}', got '{context.wired_response.error.code}'"
     )
+
+
+# Restore default step matcher for other step files
+use_step_matcher("parse")

features/steps/async_execution_steps.py

@@ -8,7 +8,7 @@ import time
 from datetime import UTC, datetime, timedelta
 from unittest.mock import MagicMock
 
-from behave import given, then, when
+from behave import given, then, use_step_matcher, when
 from behave.runner import Context
 from pydantic import ValidationError
 from ulid import ULID
@@ -45,6 +45,9 @@ PLAN_ID = "01HXYZ01HXYZ01HXYZ01HXYZ01"
 # ---------------------------------------------------------------------------
 
 
+use_step_matcher("parse")
+
+
 @given("an async job store is initialised")
 def step_async_job_store_initialised(context: Context) -> None:
     context.job_store = InMemoryJobStore()

features/steps/cli_streaming_steps.py

@@ -8,7 +8,7 @@ import time
 from pathlib import Path
 
 import parse
-from behave import given, register_type, then, when
+from behave import given, register_type, then, use_step_matcher, when
 
 from features.steps.provider_stub_utils import (
     RecordingProviderStub,
@@ -21,6 +21,7 @@ def _parse_optional_quoted(text: str) -> str:
     return text
 
 
+use_step_matcher("parse")
 register_type(OptionalQuoted=_parse_optional_quoted)
 
 

features/steps/config_cli_safety_net_coverage_steps.py

@@ -20,7 +20,7 @@ from unittest.mock import patch
 
 import typer
 import yaml
-from behave import given, then, when
+from behave import given, then, use_step_matcher, when
 from behave.runner import Context
 from rich.console import Console
 from typer.testing import CliRunner
@@ -81,6 +81,9 @@ def _teardown_safety_net_temp(context: Context) -> None:
 # ===================================================================
 
 
+use_step_matcher("parse")
+
+
 @given("a safety-net isolated temp config directory")
 def step_sn_temp_dir(context: Context) -> None:
     _setup_safety_net_temp(context)

features/steps/domain_analyzer_steps.py

@@ -399,3 +399,7 @@ def step_then_all_confidence_scores(context: Context, score: float) -> None:
         assert triple.confidence == score, (
             f"Triple {triple!r} has confidence={triple.confidence}, expected {score}"
         )
+
+
+# Restore default step matcher for other step files
+use_step_matcher("parse")

features/steps/namespaced_project_model_steps.py

@@ -611,3 +611,7 @@ def step_check_retrieved_link_none(context: Context) -> None:
 def step_check_retrieved_link_alias(context: Context, expected: str) -> None:
     """Check retrieved link alias."""
     assert context.retrieved_link.alias == expected
+
+
+# Restore default step matcher for other step files
+use_step_matcher("parse")

features/steps/output_rendering_steps.py

@@ -2738,3 +2738,7 @@ def step_find_node_wrong_root(context: Context) -> None:
     context.tree_handle._path_cache[""] = context.tree_handle._element.root
     result = context.tree_handle._find_node("WrongRoot/child")
     assert result is None, f"Expected None, got {result}"
+
+
+# Restore default step matcher for other step files
+use_step_matcher("parse")

features/steps/plan_commands_new_coverage_steps.py

@@ -248,3 +248,7 @@ def step_new_cov_pas_called_with_lifecycle(context: Context) -> None:
     context.new_cov_mock_pas_cls.assert_called_once_with(
         lifecycle_service=context.new_cov_mock_lifecycle_used,
     )
+
+
+# Restore default step matcher for other step files
+use_step_matcher("parse")

features/steps/plan_diff_artifacts_uncovered_steps.py

@@ -312,3 +312,7 @@ def step_pda_t_no_ghost(context: Context) -> None:
     assert "ghost_key =" not in context.pda_console_output, (
         f"Did not expect 'ghost_key =' in: {context.pda_console_output}"
     )
+
+
+# Restore default step matcher for other step files
+use_step_matcher("parse")

features/steps/postgresql_analyzer_coverage_boost_steps.py

@@ -202,3 +202,7 @@ def step_then_result_contains_schema_view_link(
         f"{schema!r} to view containing {view!r}.\n"
         f"Triples: {context.triples}"
     )
+
+
+# Restore default step matcher for other step files
+use_step_matcher("parse")

features/steps/server_client_stubs_steps.py

@@ -445,3 +445,7 @@ def step_resolve_server_mode(context: Context, mode: str) -> None:
             os.environ["HOME"] = old_home
         else:
             os.environ.pop("HOME", None)
+
+
+# Restore default step matcher for other step files
+use_step_matcher("parse")

features/steps/server_lifecycle_steps.py

@@ -0,0 +1,292 @@
+"""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
+
+
+@when('I try to create an ASGI application with host ""')
+def step_create_asgi_empty_host(context: Any) -> None:
+    context.caught_exception = None
+    try:
+        create_asgi_app(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 host ""')
+def step_create_lifecycle_empty_host(context: Any) -> None:
+    context.caught_exception = None
+    try:
+        ServerLifecycle(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)

features/steps/validation_pipeline_steps.py

@@ -545,3 +545,7 @@ def _find_result(summary: ValidationSummary, name: str) -> ValidationResult:
     available = [r.validation_name for r in summary.results]
     msg = f"No result for '{name}', available: {available}"
     raise AssertionError(msg)
+
+
+# Restore default step matcher for other step files
+use_step_matcher("parse")

noxfile.py

@@ -932,6 +932,8 @@ def security_scan(session: nox.Session):
       - Vulture >=80% confidence: hard fail
     """
     session.install("-e", ".[dev]")
+    # Ensure setuptools is available for semgrep's pkg_resources dependency
+    session.install("setuptools<75")
 
     # Ensure output directory exists (CI starts with a clean checkout)
     os.makedirs("build", exist_ok=True)

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
@@ -70,6 +71,7 @@ dev = [
     # Security scanning
     "bandit[toml]>=1.7.5",
     "semgrep>=1.60.0",
+    "setuptools>=70.0.0,<75.0.0",  # Required by semgrep (pkg_resources, removed in >=75)
     # Dead code detection
     "vulture>=2.10",
     # Complexity metrics

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()

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

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",
 ]

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,

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",
+]

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",
+]

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",
+]

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