fix: enforce atomic agent registration contract

This commit is contained in:
ZacharyZhang-NY committed 2026-07-19 15:38:40 -04:00
1 parent 47d323c17b
commit f64d00673c
58 files changed
+2692 -1667

No files matched your search

+4
View File
@@ -44,6 +44,8 @@ from .types import (
GetMeResponse,
IssueTokenResponse,
HealthResponse,
INTEGRATION_TYPES,
IntegrationType,
)
__version__ = "1.0.0"
@@ -93,4 +95,6 @@ __all__ = [
"GetMeResponse",
"IssueTokenResponse",
"HealthResponse",
"INTEGRATION_TYPES",
"IntegrationType",
]
+2
View File
@@ -11,6 +11,7 @@ import aiohttp
from .crypto import compute_chain_hash, compute_payload_hash, sign_eor
from .errors import ElydoraError
from .integration_types import require_integration_type
from .types import (
AuditQueryResponse,
AuthLoginResponse,
@@ -231,6 +232,7 @@ class AsyncElydoraClient:
async def register_agent(self, request: RegisterAgentRequest) -> RegisterAgentResponse:
"""Register a new agent with the organization."""
require_integration_type(request.get("integration_type"))
return await self._request("POST", "/v1/agents/register", json_body=request)
async def get_agent(self, agent_id: str) -> GetAgentResponse:
+2
View File
@@ -10,6 +10,7 @@ import requests
from .crypto import compute_chain_hash, compute_payload_hash, sign_eor
from .errors import ElydoraError
from .integration_types import require_integration_type
from .types import (
AuditQueryResponse,
AuthLoginResponse,
@@ -222,6 +223,7 @@ class ElydoraClient:
def register_agent(self, request: RegisterAgentRequest) -> RegisterAgentResponse:
"""Register a new agent with the organization."""
require_integration_type(request.get("integration_type"))
return self._request("POST", "/v1/agents/register", json_body=request)
def get_agent(self, agent_id: str) -> GetAgentResponse:
+12
View File
@@ -0,0 +1,12 @@
"""Runtime validation for agent integration types."""
from typing import cast
from .types import INTEGRATION_TYPES, IntegrationType
def require_integration_type(value: object) -> IntegrationType:
if not isinstance(value, str) or value not in INTEGRATION_TYPES:
expected = ", ".join(INTEGRATION_TYPES)
raise ValueError(f"Invalid integration_type {value!r}. Expected one of: {expected}")
return cast(IntegrationType, value)
+34 -8
View File
@@ -3,15 +3,13 @@
All types use TypedDict for structural typing, compatible with Python 3.9+.
"""
from __future__ import annotations
import sys
from typing import Any, Dict, List, Optional
from typing import Any, Dict, List, Optional, Tuple
if sys.version_info >= (3, 11):
from typing import Literal, TypedDict
from typing import Literal, NotRequired, TypedDict
else:
from typing_extensions import Literal, TypedDict
from typing_extensions import Literal, NotRequired, TypedDict
# ---------------------------------------------------------------------------
@@ -28,6 +26,32 @@ RbacRole = Literal[
"readonly_investigator",
"integration_engineer",
]
IntegrationType = Literal[
"augment",
"claudecode",
"cline",
"codex",
"copilot",
"cursor",
"droid",
"gemini",
"grok",
"kimi",
"kirocli",
"kiroide",
"letta",
"opencode",
"qwen",
"enterprise",
"gui",
"sdk",
"other",
]
INTEGRATION_TYPES: Tuple[IntegrationType, ...] = (
"augment", "claudecode", "cline", "codex", "copilot", "cursor", "droid",
"gemini", "grok", "kimi", "kirocli", "kiroide", "letta", "opencode", "qwen",
"enterprise", "gui", "sdk", "other",
)
ErrorCode = Literal[
"INVALID_SIGNATURE",
"UNKNOWN_AGENT",
@@ -55,6 +79,7 @@ class Agent(TypedDict):
org_id: str
display_name: str
responsible_entity: str
integration_type: IntegrationType
status: AgentStatus
created_at: int
updated_at: int
@@ -187,11 +212,12 @@ class RegisterAgentKeyParam(TypedDict):
algorithm: Literal["ed25519"]
class RegisterAgentRequest(TypedDict, total=False):
class RegisterAgentRequest(TypedDict):
agent_id: str
display_name: str
responsible_entity: str
integration_type: IntegrationType
keys: List[RegisterAgentKeyParam]
display_name: NotRequired[str]
responsible_entity: NotRequired[str]
class AuditQueryParams(TypedDict, total=False):