feat: expose auth admin summaries

This commit is contained in:
lda
2026-06-06 12:16:45 +07:00 Verified
parent 58771e4423
commit c25c4cfee5
18 changed files with 478 additions and 8 deletions
+2
View File
@@ -4,6 +4,7 @@ from .auth import AUTH_ID_PATTERN, AuthRecord, AuthStore, validate_auth_id
from .listing import matches_query, paged_list_payload
from .admin import (
WorkflowAdminApi,
WorkflowAdminAuthProvider,
WorkflowAdminConnectionProvider,
WorkflowAdminEventProvider,
)
@@ -91,6 +92,7 @@ __all__ = [
"RuntimeDependencies",
"TraceRange",
"WorkflowAdminApi",
"WorkflowAdminAuthProvider",
"WorkflowAdminConnectionProvider",
"WorkflowAdminEventProvider",
"WorkflowAdminSurface",
+24
View File
@@ -19,6 +19,14 @@ class WorkflowAdminEventProvider(Protocol):
def list_events(self) -> Sequence[Mapping[str, Any] | object]: ...
class WorkflowAdminAuthProvider(Protocol):
"""Provides read-only auth inventory without secret payload values."""
def list_auth_records(self) -> Sequence[Mapping[str, Any] | object]: ...
def inspect_auth_record(self, auth_ref: str) -> Mapping[str, Any] | object: ...
class WorkflowAdminApi:
"""Protocol-neutral read-only broker/server admin operations.
@@ -31,9 +39,11 @@ class WorkflowAdminApi:
*,
connections: WorkflowAdminConnectionProvider,
events: WorkflowAdminEventProvider,
auth: WorkflowAdminAuthProvider | None = None,
) -> None:
self.connections = connections
self.events = events
self.auth = auth
async def list_connections(self) -> dict[str, Any]:
connections = sorted(
@@ -55,6 +65,20 @@ class WorkflowAdminApi:
events = [_payload(event) for event in self.events.list_events()]
return {"events": events, "total": len(events)}
async def list_auth_records(self) -> dict[str, Any]:
if self.auth is None:
raise RuntimeError("auth admin is not available for this target")
records = sorted(
(_payload(item) for item in self.auth.list_auth_records()),
key=lambda item: str(item.get("id", "")),
)
return {"auth_records": records, "total": len(records)}
async def inspect_auth_record(self, auth_ref: str) -> dict[str, Any]:
if self.auth is None:
raise RuntimeError("auth admin is not available for this target")
return _payload(self.auth.inspect_auth_record(auth_ref))
def _payload(value: Mapping[str, Any] | object) -> dict[str, Any]:
"""Normalize provider objects without depending on MCP event/config types."""
+4
View File
@@ -225,6 +225,10 @@ class WorkflowAdminSurface(Protocol):
async def list_events(self) -> dict[str, Any]: ...
async def list_auth_records(self) -> dict[str, Any]: ...
async def inspect_auth_record(self, auth_ref: str) -> dict[str, Any]: ...
class WorkflowSourceRegistrySurface(Protocol):
"""Desired source registry methods exposed by platform frontends."""