fix(sdk): enforce agent registration integration type

This commit is contained in:
ZacharyZhang-NY committed 2026-07-19 14:59:34 -04:00
1 parent ecc70adf12
commit c000855ebc
8 files changed
+200 -20

No files matched your search

+2
View File
@@ -32,6 +32,7 @@ from .types import (
AdminAction,
AgentStatus,
ExportStatus,
INTEGRATION_TYPES,
IntegrationType,
KeyStatus,
RbacRole,
@@ -90,6 +91,7 @@ __all__ = [
"AdminAction",
"AgentStatus",
"ExportStatus",
"INTEGRATION_TYPES",
"IntegrationType",
"KeyStatus",
"RbacRole",
+5 -1
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,
@@ -27,6 +28,7 @@ from .types import (
GetOperationResponse,
HealthResponse,
IssueApiTokenResponse,
IntegrationType,
ListAdminEventsResponse,
ListMembersResponse,
ListWebhooksResponse,
@@ -248,6 +250,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:
@@ -266,8 +269,9 @@ class AsyncElydoraClient:
"""Unfreeze an agent."""
return await self._request("POST", f"/v1/agents/{agent_id}/unfreeze", json_body={"reason": reason})
async def update_agent(self, agent_id: str, integration_type: str) -> UpdateAgentResponse:
async def update_agent(self, agent_id: str, integration_type: IntegrationType) -> UpdateAgentResponse:
"""Update an agent's integration type."""
require_integration_type(integration_type)
return await self._request("PATCH", f"/v1/agents/{agent_id}", json_body={"integration_type": integration_type})
async def delete_agent(self, agent_id: str) -> DeleteAgentResponse:
+5 -1
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,
@@ -26,6 +27,7 @@ from .types import (
GetOperationResponse,
HealthResponse,
IssueApiTokenResponse,
IntegrationType,
ListAdminEventsResponse,
ListMembersResponse,
ListWebhooksResponse,
@@ -239,6 +241,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:
@@ -257,8 +260,9 @@ class ElydoraClient:
"""Unfreeze an agent."""
return self._request("POST", f"/v1/agents/{agent_id}/unfreeze", json_body={"reason": reason})
def update_agent(self, agent_id: str, integration_type: str) -> UpdateAgentResponse:
def update_agent(self, agent_id: str, integration_type: IntegrationType) -> UpdateAgentResponse:
"""Update an agent's integration type."""
require_integration_type(integration_type)
return self._request("PATCH", f"/v1/agents/{agent_id}", json_body={"integration_type": integration_type})
def delete_agent(self, agent_id: str) -> DeleteAgentResponse:
+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)
+33 -18
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
# ---------------------------------------------------------------------------
@@ -29,15 +27,31 @@ RbacRole = Literal[
"integration_engineer",
]
IntegrationType = Literal[
"openai",
"anthropic",
"augment",
"claudecode",
"cline",
"codex",
"copilot",
"cursor",
"droid",
"gemini",
"langchain",
"llamaindex",
"autogen",
"crewai",
"custom",
"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",
)
AdminAction = Literal[
"agent.freeze",
"agent.unfreeze",
@@ -78,7 +92,7 @@ class Agent(TypedDict):
display_name: str
responsible_entity: str
status: AgentStatus
integration_type: str
integration_type: IntegrationType
created_at: int
updated_at: int
@@ -212,11 +226,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):
@@ -410,8 +425,8 @@ class MemberEntry(TypedDict):
# ---------------------------------------------------------------------------
class UpdateAgentRequest(TypedDict, total=False):
integration_type: str
class UpdateAgentRequest(TypedDict):
integration_type: IntegrationType
# ---------------------------------------------------------------------------