mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-20 15:33:54 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

feat(gateway): route platform HTTP event callbacks

Browse source
This commit is contained in:
aeyeopsdev 2026-05-31 19:36:32 +00:00 • committed by Teknium
parent 03c0b00f45
commit 1305a690e0
6 changed files with 420 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

122
gateway/platforms/api_server.py
View file

@ -995,6 +995,10 @@ class APIServerAdapter(BasePlatformAdapter):
# (the /v1/runs path tracks its own in-flight set via
# _active_run_tasks).
self._inflight_agent_runs: int = 0
# Back-reference to the owning GatewayRunner (set by gateway/run.py)
# so /api/platforms/{platform}/events can resolve sibling adapters.
# BasePlatformAdapter declares the class-level default of None.
self.gateway_runner: Optional[Any] = None
# Requests admitted before their handler reaches agent bookkeeping.
# Shutdown counts this reservation so the request cannot slip through
# the drain between its first await and _run_agent()/task registration.
@ -1254,6 +1258,118 @@ class APIServerAdapter(BasePlatformAdapter):
status=401,
)
@staticmethod
def _normalize_callback_platform(value: str) -> str:
normalized = (value or "").strip().lower().replace("-", "_")
if not re.fullmatch(r"[a-z0-9_]+", normalized):
return ""
return normalized
def _get_platform_callback_adapter(
self,
request: "web.Request",
platform_name: str,
) -> Optional[Any]:
injected = request.app.get("platform_event_adapters")
if isinstance(injected, dict):
adapter = injected.get(platform_name)
if adapter is not None:
return adapter
adapter = request.app.get(f"{platform_name}_adapter")
if adapter is not None:
return adapter
runner = self.gateway_runner or request.app.get("gateway_runner")
adapters = getattr(runner, "adapters", None)
if not adapters:
return None
try:
from gateway.config import Platform as _Platform
return adapters.get(_Platform(platform_name))
except Exception:
for platform, candidate in adapters.items():
if getattr(platform, "value", platform) == platform_name:
return candidate
return None
async def _handle_platform_event_callback(self, request: "web.Request") -> "web.Response":
platform_name = self._normalize_callback_platform(
request.match_info.get("platform", "")
)
if not platform_name:
return web.json_response(
_openai_error(
"Invalid platform name",
code="invalid_platform",
),
status=400,
)
adapter = self._get_platform_callback_adapter(request, platform_name)
if adapter is None:
return web.json_response(
_openai_error(
"Platform adapter is not connected",
code="platform_unavailable",
),
status=503,
)
verifier = getattr(adapter, "verify_http_event_request", None)
dispatcher = getattr(adapter, "dispatch_http_event", None)
if verifier is None or dispatcher is None:
return web.json_response(
_openai_error(
"Platform adapter does not support HTTP events",
code="platform_http_events_unsupported",
),
status=503,
)
ok, code = verifier(request.headers.get("Authorization", ""))
if not ok:
return web.json_response(
_openai_error(
"Invalid platform event authorization",
code=code or "invalid_platform_event_authorization",
),
status=401,
)
try:
payload = await request.json()
except Exception:
return web.json_response(
_openai_error("Invalid JSON in platform event", code="invalid_json"),
status=400,
)
if not isinstance(payload, dict):
return web.json_response(
_openai_error(
"Platform event must be a JSON object",
code="invalid_request",
),
status=400,
)
try:
result = await dispatcher(payload)
except Exception:
logger.exception("Platform HTTP event dispatch failed for %s", platform_name)
return web.json_response(
_openai_error(
"Platform event dispatch failed",
err_type="server_error",
code="platform_event_dispatch_failed",
),
status=500,
)
return web.json_response(result if isinstance(result, dict) else {})
# ------------------------------------------------------------------
# Multi-profile multiplexing (/p/<profile>/…)
# ------------------------------------------------------------------
@ -1364,6 +1480,10 @@ class APIServerAdapter(BasePlatformAdapter):
("POST", "/v1/responses", self._handle_responses),
("GET", "/v1/responses/{response_id}", self._handle_get_response),
("DELETE", "/v1/responses/{response_id}", self._handle_delete_response),
# Generic platform HTTP event callback ingress. Authenticated by
# the target adapter's own verifier (platform-signed bearer), NOT
# API_SERVER_KEY — external platforms hold no API server key.
("POST", "/api/platforms/{platform}/events", self._handle_platform_event_callback),
("GET", "/api/jobs", self._handle_list_jobs),
("POST", "/api/jobs", self._handle_create_job),
("GET", "/api/jobs/{job_id}", self._handle_get_job),
@ -5174,6 +5294,8 @@ class APIServerAdapter(BasePlatformAdapter):
# native routes first lets those shims no-op instead of shadowing the
# upstream session-control handlers.
self._app["api_server_adapter"] = self
if self.gateway_runner is not None:
self._app["gateway_runner"] = self.gateway_runner
# Start background sweep to clean up orphaned (unconsumed) run streams
sweep_task = asyncio.create_task(self._sweep_orphaned_runs())

4
gateway/run.py
View file

@ -9042,7 +9042,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
if not check_api_server_requirements():
logger.warning("API Server: aiohttp not installed")
return None
return APIServerAdapter(config)
adapter = APIServerAdapter(config)
adapter.gateway_runner = self
return adapter
elif platform == Platform.WEBHOOK:
from gateway.platforms.webhook import WebhookAdapter, check_webhook_requirements

91
plugins/platforms/google_chat/adapter.py
View file

@ -73,6 +73,20 @@ MediaFileUpload: Any = None # type: ignore
_google_modules_loaded: bool = False
def _verify_google_id_token(token: str, audience: str) -> Dict[str, Any]:
try:
from google.auth.transport import requests as google_requests
from google.oauth2 import id_token
except ImportError as exc:
raise RuntimeError("google-auth is required for Google Chat HTTP callbacks") from exc
return id_token.verify_oauth2_token(
token,
google_requests.Request(),
audience,
)
def _load_google_modules() -> bool:
"""Lazily import the heavy google-cloud + googleapiclient stack.
@ -690,6 +704,21 @@ class GoogleChatAdapter(BasePlatformAdapter):
self._max_bytes = int(os.getenv("GOOGLE_CHAT_MAX_BYTES", str(16 * 1024 * 1024)))
except (ValueError, TypeError):
self._max_bytes = 16 * 1024 * 1024
self._http_events_url = (
self.config.extra.get("http_events_url")
or os.getenv("GOOGLE_CHAT_HTTP_EVENTS_URL", "")
or ""
).strip()
self._http_events_audience = (
self.config.extra.get("http_events_audience")
or os.getenv("GOOGLE_CHAT_HTTP_EVENTS_AUDIENCE", "")
or self._http_events_url
).strip()
self._http_events_service_account_email = (
self.config.extra.get("http_events_service_account_email")
or os.getenv("GOOGLE_CHAT_HTTP_EVENTS_SERVICE_ACCOUNT_EMAIL", "")
or ""
).strip().lower()
# ------------------------------------------------------------------
# Configuration loading and validation
@ -1422,6 +1451,62 @@ class GoogleChatAdapter(BasePlatformAdapter):
except Exception:
pass
async def dispatch_http_event(self, envelope: Dict[str, Any]) -> Dict[str, Any]:
extracted = self._extract_message_payload(envelope)
if extracted is None:
return {}
msg, space, _fmt = extracted
sender = msg.get("sender") or {}
if sender.get("type") == "BOT":
return {}
msg_name = msg.get("name") or ""
if msg_name and self._dedup.is_duplicate(msg_name):
return {}
msg_with_space = dict(msg)
if "space" not in msg_with_space and space:
msg_with_space["space"] = space
enriched_env = dict(envelope)
if "space" not in enriched_env and space:
enriched_env["space"] = space
await self._dispatch_message(msg_with_space, enriched_env)
return {}
def verify_http_event_request(self, auth_header: str) -> Tuple[bool, str]:
if not self._http_events_audience or not self._http_events_service_account_email:
return False, "google_chat_http_events_not_configured"
if not auth_header.startswith("Bearer "):
return False, "missing_google_bearer"
token = auth_header[7:].strip()
if not token:
return False, "missing_google_bearer"
try:
claims = _verify_google_id_token(token, self._http_events_audience)
except Exception as exc:
logger.warning(
"[GoogleChat] HTTP event bearer verification failed: %s",
_redact_sensitive(str(exc)),
)
return False, "invalid_google_bearer"
expected = {
item.strip().lower()
for item in self._http_events_service_account_email.split(",")
if item.strip()
}
claim_email = str(claims.get("email") or "").strip().lower()
if not claim_email or claim_email not in expected:
return False, "unexpected_google_bearer_identity"
return True, ""
async def _dispatch_message(self, msg: Dict[str, Any], envelope: Dict[str, Any]) -> None:
"""Translate a Chat message payload to a MessageEvent and hand off.
@ -3270,6 +3355,12 @@ def _env_enablement() -> Optional[Dict[str, Any]]:
seed["subscription_name"] = subscription
if http_events_url:
seed["http_events_url"] = http_events_url
http_events_audience = os.getenv("GOOGLE_CHAT_HTTP_EVENTS_AUDIENCE")
if http_events_audience:
seed["http_events_audience"] = http_events_audience
http_events_sa_email = os.getenv("GOOGLE_CHAT_HTTP_EVENTS_SERVICE_ACCOUNT_EMAIL")
if http_events_sa_email:
seed["http_events_service_account_email"] = http_events_sa_email
sa_json = (
os.getenv("GOOGLE_CHAT_SERVICE_ACCOUNT_JSON")
or os.getenv("GOOGLE_APPLICATION_CREDENTIALS")

8
plugins/platforms/google_chat/plugin.yaml
View file

@ -23,6 +23,14 @@ optional_env:
description: "Authenticated HTTP endpoint for Chat message events."
prompt: "HTTP events callback URL"
password: false
- name: GOOGLE_CHAT_HTTP_EVENTS_AUDIENCE
description: "Expected audience for Google-signed HTTP event bearer tokens. Defaults to GOOGLE_CHAT_HTTP_EVENTS_URL."
prompt: "HTTP events token audience"
password: false
- name: GOOGLE_CHAT_HTTP_EVENTS_SERVICE_ACCOUNT_EMAIL
description: "Expected Google service account email for HTTP event bearer tokens."
prompt: "HTTP events service account email"
password: false
- name: GOOGLE_CHAT_PROJECT_ID
description: "GCP project ID for optional Pub/Sub inbound mode. Falls back to GOOGLE_CLOUD_PROJECT."
prompt: "GCP project ID"

94
tests/gateway/test_api_server.py
View file

@ -652,9 +652,28 @@ def _create_app(adapter: APIServerAdapter) -> web.Application:
app.router.add_post("/v1/responses", adapter._handle_responses)
app.router.add_get("/v1/responses/{response_id}", adapter._handle_get_response)
app.router.add_delete("/v1/responses/{response_id}", adapter._handle_delete_response)
app.router.add_post(
"/api/platforms/{platform}/events",
adapter._handle_platform_event_callback,
)
return app
class _FakeGoogleChatAdapter:
def __init__(self, *, verify_ok: bool = True, verify_code: str = ""):
self.verify_ok = verify_ok
self.verify_code = verify_code
self.dispatched = []
def verify_http_event_request(self, auth_header: str):
self.auth_header = auth_header
return self.verify_ok, self.verify_code
async def dispatch_http_event(self, payload):
self.dispatched.append(payload)
return {"ok": True}
@pytest.fixture
def adapter():
return _make_adapter()
@ -2850,6 +2869,81 @@ class TestSendMethod:
assert "HTTP request/response" in result.error
class TestPlatformEventCallbackEndpoint:
@pytest.mark.asyncio
async def test_dispatches_authorized_google_chat_event(self, adapter):
app = _create_app(adapter)
google_adapter = _FakeGoogleChatAdapter()
app["platform_event_adapters"] = {"google_chat": google_adapter}
async with TestClient(TestServer(app)) as cli:
resp = await cli.post(
"/api/platforms/google_chat/events",
headers={"Authorization": "Bearer google-token"},
json={"type": "MESSAGE", "message": {"text": "hi"}},
)
body = await resp.json()
assert resp.status == 200
assert body == {"ok": True}
assert google_adapter.auth_header == "Bearer google-token"
assert google_adapter.dispatched == [
{"type": "MESSAGE", "message": {"text": "hi"}}
]
@pytest.mark.asyncio
async def test_rejects_invalid_google_chat_auth(self, adapter):
app = _create_app(adapter)
app["platform_event_adapters"] = {
"google_chat": _FakeGoogleChatAdapter(
verify_ok=False,
verify_code="invalid_google_bearer",
)
}
async with TestClient(TestServer(app)) as cli:
resp = await cli.post(
"/api/platforms/google_chat/events",
headers={"Authorization": "Bearer bad"},
json={"type": "MESSAGE"},
)
body = await resp.json()
assert resp.status == 401
assert body["error"]["code"] == "invalid_google_bearer"
@pytest.mark.asyncio
async def test_requires_connected_google_chat_adapter(self, adapter):
app = _create_app(adapter)
async with TestClient(TestServer(app)) as cli:
resp = await cli.post(
"/api/platforms/google_chat/events",
headers={"Authorization": "Bearer google-token"},
json={"type": "MESSAGE"},
)
body = await resp.json()
assert resp.status == 503
assert body["error"]["code"] == "platform_unavailable"
@pytest.mark.asyncio
async def test_rejects_malformed_platform_event_json(self, adapter):
app = _create_app(adapter)
app["platform_event_adapters"] = {"google_chat": _FakeGoogleChatAdapter()}
async with TestClient(TestServer(app)) as cli:
resp = await cli.post(
"/api/platforms/google_chat/events",
headers={"Authorization": "Bearer google-token"},
data="{",
)
body = await resp.json()
assert resp.status == 400
assert body["error"]["code"] == "invalid_json"
# ---------------------------------------------------------------------------
# GET /v1/responses/{response_id}
# ---------------------------------------------------------------------------

104
tests/gateway/test_google_chat.py
View file

@ -254,6 +254,8 @@ class TestEnvConfigLoading:
"GOOGLE_CHAT_SUBSCRIPTION_NAME",
"GOOGLE_CHAT_SUBSCRIPTION",
"GOOGLE_CHAT_HTTP_EVENTS_URL",
"GOOGLE_CHAT_HTTP_EVENTS_AUDIENCE",
"GOOGLE_CHAT_HTTP_EVENTS_SERVICE_ACCOUNT_EMAIL",
"GOOGLE_CHAT_SERVICE_ACCOUNT_JSON",
"GOOGLE_APPLICATION_CREDENTIALS",
"GOOGLE_CHAT_HOME_CHANNEL",
@ -284,10 +286,32 @@ class TestEnvConfigLoading:
def test_http_events_enable_without_pubsub(self, monkeypatch):
self._clean_env(monkeypatch)
monkeypatch.setenv("GOOGLE_CHAT_HTTP_EVENTS_URL", "https://example.test/google-chat/events")
monkeypatch.setenv(
"GOOGLE_CHAT_HTTP_EVENTS_URL",
"https://example.test/google-chat/events",
)
monkeypatch.setenv(
"GOOGLE_CHAT_HTTP_EVENTS_AUDIENCE",
"https://callback.example.test/events",
)
monkeypatch.setenv(
"GOOGLE_CHAT_HTTP_EVENTS_SERVICE_ACCOUNT_EMAIL",
"chat-callback@example.iam.gserviceaccount.com",
)
cfg = load_gateway_config()
assert _GC in cfg.platforms
assert cfg.platforms[_GC].extra["http_events_url"] == "https://example.test/google-chat/events"
assert (
cfg.platforms[_GC].extra["http_events_url"]
== "https://example.test/google-chat/events"
)
assert (
cfg.platforms[_GC].extra["http_events_audience"]
== "https://callback.example.test/events"
)
assert (
cfg.platforms[_GC].extra["http_events_service_account_email"]
== "chat-callback@example.iam.gserviceaccount.com"
)
# ===========================================================================
@ -414,6 +438,82 @@ class TestValidateConfig:
assert sub == "projects/inferred/subscriptions/sub"
class TestHttpEventIngress:
def test_verify_http_event_request_accepts_expected_google_identity(self, monkeypatch):
cfg = PlatformConfig(enabled=True)
cfg.extra.update(
{
"http_events_url": "https://example.test/google-chat/events",
"http_events_service_account_email": (
"chat-callback@example.iam.gserviceaccount.com"
),
}
)
a = GoogleChatAdapter(cfg)
def fake_verify(token, audience):
assert token == "signed-token"
assert audience == "https://example.test/google-chat/events"
return {"email": "chat-callback@example.iam.gserviceaccount.com"}
monkeypatch.setattr(_gc_mod, "_verify_google_id_token", fake_verify)
assert a.verify_http_event_request("Bearer signed-token") == (True, "")
def test_verify_http_event_request_rejects_unexpected_identity(self, monkeypatch):
cfg = PlatformConfig(enabled=True)
cfg.extra.update(
{
"http_events_url": "https://example.test/google-chat/events",
"http_events_service_account_email": (
"expected@example.iam.gserviceaccount.com"
),
}
)
a = GoogleChatAdapter(cfg)
monkeypatch.setattr(
_gc_mod,
"_verify_google_id_token",
lambda _token, _audience: {
"email": "other@example.iam.gserviceaccount.com"
},
)
ok, code = a.verify_http_event_request("Bearer signed-token")
assert ok is False
assert code == "unexpected_google_bearer_identity"
def test_verify_http_event_request_requires_callback_identity_config(self):
cfg = PlatformConfig(enabled=True)
cfg.extra["http_events_url"] = "https://example.test/google-chat/events"
a = GoogleChatAdapter(cfg)
assert a.verify_http_event_request("Bearer signed-token") == (
False,
"google_chat_http_events_not_configured",
)
@pytest.mark.asyncio
async def test_dispatch_http_event_routes_message_payload(self, adapter):
envelope = _make_chat_envelope(text="hello from http")
result = await adapter.dispatch_http_event(envelope)
assert result == {}
adapter.handle_message.assert_awaited_once()
event = adapter.handle_message.await_args.args[0]
assert event.text == "hello from http"
assert event.source.chat_id == "spaces/S"
@pytest.mark.asyncio
async def test_dispatch_http_event_ignores_bot_messages(self, adapter):
envelope = _make_chat_envelope(text="bot echo", sender_type="BOT")
assert await adapter.dispatch_http_event(envelope) == {}
adapter.handle_message.assert_not_awaited()
class TestConnectModes:
@pytest.mark.asyncio
async def test_connect_http_mode_skips_pubsub_subscriber(self, tmp_path, monkeypatch):