forked from cleveragents/cleveragents-core
freemo
5f0a545113
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
118 lines
3.4 KiB
Python
118 lines
3.4 KiB
Python
#!/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()
|