Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions models/src/agent_control_models/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,9 @@
from .server import (
AgentRef,
AgentSummary,
CloneAndBindControlRequest,
CloneAndBindControlResponse,
CloneAndBindTargetBinding,
ConflictMode,
ControlSummary,
ControlVersionSummary,
Expand Down Expand Up @@ -176,6 +179,9 @@
# Server models
"AgentRef",
"AgentSummary",
"CloneAndBindControlRequest",
"CloneAndBindControlResponse",
"CloneAndBindTargetBinding",
"ConflictMode",
"ControlVersionSummary",
"ControlSummary",
Expand Down
51 changes: 49 additions & 2 deletions models/src/agent_control_models/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -347,6 +347,9 @@ class GetControlResponse(BaseModel):

id: int = Field(..., description="Control ID")
name: str = Field(..., description="Control name")
cloned_from_control_id: int | None = Field(
None, description="Source control ID when this control is a clone."
)
data: ControlDefinition | UnrenderedTemplateControl = Field(
description=(
"Control configuration data. A ControlDefinition for raw/rendered "
Expand Down Expand Up @@ -519,6 +522,9 @@ class ControlSummary(BaseModel):

id: int = Field(..., description="Control ID")
name: str = Field(..., description="Control name")
cloned_from_control_id: int | None = Field(
None, description="Source control ID when this control is a clone."
)
description: str | None = Field(None, description="Control description")
enabled: bool = Field(True, description="Whether control is enabled")
execution: str | None = Field(None, description="'server' or 'sdk'")
Expand Down Expand Up @@ -580,7 +586,7 @@ class GetControlVersionResponse(BaseModel):
...,
description=(
"Raw persisted snapshot of the control state at this version, including "
"metadata such as name, deleted_at, and cloned_control_id."
"metadata such as name, deleted_at, and cloned_from_control_id."
),
)

Expand Down Expand Up @@ -635,6 +641,48 @@ class PatchControlResponse(BaseModel):
]


class CloneAndBindTargetBinding(BaseModel):
"""Target binding to create for a cloned control."""

target_type: ControlBindingTargetField = Field(
...,
description="Opaque attachment kind (caller-defined; e.g. 'environment', 'session').",
)
target_id: ControlBindingTargetField = Field(
..., description="Opaque external identifier within the target_type."
)
enabled: bool = Field(
default=True,
description="Whether the created binding is active.",
)


class CloneAndBindControlRequest(BaseModel):
"""Request to clone a control and attach the clone to one target."""

model_config = ConfigDict(extra="forbid")

name: SlugName | None = Field(
None,
description=(
"Optional unique name for the cloned control. If omitted, the server "
"generates a name from the source control name."
),
)
target_binding: CloneAndBindTargetBinding = Field(
..., description="Target binding to create for the cloned control."
)


class CloneAndBindControlResponse(BaseModel):
"""Response from cloning and binding a control."""

control_id: int = Field(..., description="Identifier of the cloned control.")
name: str = Field(..., description="Name of the cloned control.")
cloned_from_control_id: int = Field(..., description="Source control ID.")
binding_id: int = Field(..., description="Identifier of the created binding.")


class CreateControlBindingRequest(BaseModel):
"""Request to attach a control to an opaque external target."""

Expand Down Expand Up @@ -759,4 +807,3 @@ class DeleteControlBindingByKeyResponse(BaseModel):
"binding existed."
),
)

47 changes: 47 additions & 0 deletions sdks/python/src/agent_control/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -1019,6 +1019,7 @@ async def list_controls(
name: str | None = None,
enabled: bool | None = None,
template_backed: bool | None = None,
cloned: bool | None = None,
step_type: str | None = None,
stage: Literal["pre", "post"] | None = None,
execution: Literal["server", "sdk"] | None = None,
Expand All @@ -1035,6 +1036,7 @@ async def list_controls(
name: Optional filter by name (partial, case-insensitive)
enabled: Optional filter by enabled status
template_backed: Optional filter by whether the control is template-backed
cloned: Optional filter by whether the control was cloned from another control
step_type: Optional filter by step type (built-ins: 'tool', 'llm')
stage: Optional filter by stage ('pre' or 'post')
execution: Optional filter by execution ('server' or 'sdk')
Expand Down Expand Up @@ -1079,6 +1081,7 @@ async def main():
name=name,
enabled=enabled,
template_backed=template_backed,
cloned=cloned,
step_type=step_type,
stage=stage,
execution=execution,
Expand Down Expand Up @@ -1147,6 +1150,49 @@ async def main():
return await controls.create_control(client, name, data=data)


async def clone_and_bind_control(
control_id: int,
*,
target_type: str,
target_id: str,
name: str | None = None,
enabled: bool = True,
server_url: str | None = None,
api_key: str | None = None,
api_key_header: str | None = None,
) -> dict[str, Any]:
"""
Clone an existing control and bind the clone to a target.

Args:
control_id: Source control ID to clone
target_type: Opaque attachment kind
target_id: Opaque external target identifier
name: Optional unique name for the cloned control
enabled: Whether the created binding is active
server_url: Optional server URL (defaults to AGENT_CONTROL_URL env var)
api_key: Optional API key for authentication (defaults to AGENT_CONTROL_API_KEY env var)

Returns:
Dictionary containing control_id, name, cloned_from_control_id, and binding_id.
"""
_final_server_url = server_url or os.getenv('AGENT_CONTROL_URL') or 'http://localhost:8000'

async with _ad_hoc_client(
server_url=_final_server_url,
api_key=api_key,
api_key_header=api_key_header,
) as client:
return await controls.clone_and_bind_control(
client,
control_id,
target_type=target_type,
target_id=target_id,
name=name,
enabled=enabled,
)


async def validate_control_data(
data: dict[str, Any] | ControlDefinition | TemplateControlInput,
server_url: str | None = None,
Expand Down Expand Up @@ -1502,6 +1548,7 @@ async def main():
"add_agent_control",
"remove_agent_control",
# Control management
"clone_and_bind_control",
"create_control",
"list_controls",
"get_control",
Expand Down
45 changes: 45 additions & 0 deletions sdks/python/src/agent_control/controls.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ async def list_controls(
name: str | None = None,
enabled: bool | None = None,
template_backed: bool | None = None,
cloned: bool | None = None,
step_type: str | None = None,
stage: Literal["pre", "post"] | None = None,
execution: Literal["server", "sdk"] | None = None,
Expand All @@ -37,6 +38,7 @@ async def list_controls(
name: Optional filter by name (partial, case-insensitive match)
enabled: Optional filter by enabled status
template_backed: Optional filter by whether the control is template-backed
cloned: Optional filter by whether the control was cloned from another control
step_type: Optional filter by step type (built-ins: 'tool', 'llm')
stage: Optional filter by stage ('pre' or 'post')
execution: Optional filter by execution ('server' or 'sdk')
Expand Down Expand Up @@ -78,6 +80,8 @@ async def list_controls(
params["enabled"] = enabled
if template_backed is not None:
params["template_backed"] = template_backed
if cloned is not None:
params["cloned"] = cloned
if step_type is not None:
params["step_type"] = step_type
if stage is not None:
Expand Down Expand Up @@ -243,6 +247,47 @@ async def create_control(
return result


async def clone_and_bind_control(
client: AgentControlClient,
control_id: int,
*,
target_type: str,
target_id: str,
name: str | None = None,
enabled: bool = True,
) -> dict[str, Any]:
"""
Clone an existing control and bind the clone to a target in one API call.

Args:
client: AgentControlClient instance
control_id: Source control ID to clone
target_type: Opaque attachment kind
target_id: Opaque external target identifier
name: Optional unique name for the cloned control
enabled: Whether the created binding is active

Returns:
Dictionary containing control_id, name, cloned_from_control_id, and binding_id.
"""
payload: dict[str, Any] = {
"target_binding": {
"target_type": target_type,
"target_id": target_id,
"enabled": enabled,
}
}
if name is not None:
payload["name"] = name

response = await client.http_client.post(
f"/api/v1/controls/{control_id}/clone-and-bind",
json=payload,
)
response.raise_for_status()
return cast(dict[str, Any], response.json())


async def set_control_data(
client: AgentControlClient,
control_id: int,
Expand Down
Loading
Loading