Skip to content
Merged
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
35 changes: 35 additions & 0 deletions engine/tests/test_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -1085,6 +1085,41 @@ def register_payload_evaluator(self):
except ValueError:
pass

@pytest.mark.asyncio
async def test_custom_step_type_filters_controls(self):
"""Controls match custom step types without falling back to LLM."""
controls = [
make_control(
1,
"retriever-control",
"test-deny",
action="deny",
config_value="retriever",
step_types=["retriever"],
),
make_control(
2,
"llm-control",
"test-deny",
action="deny",
config_value="llm",
step_types=["llm"],
),
]
engine = ControlEngine(controls)

result = await engine.process(
EvaluationRequest(
agent_name="00000000-0000-0000-0000-000000000001",
step=Step(type="retriever", name="retrieve", input="query"),
stage="pre",
)
)

assert [match.control_name for match in result.matches or []] == [
"retriever-control"
]

@pytest.mark.asyncio
async def test_step_names_filters_tasks(self):
# Given: two controls scoped to different steps
Expand Down
14 changes: 8 additions & 6 deletions models/src/agent_control_models/observability.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ class ControlExecutionEvent(BaseModel):
control_id: Database ID of the control
control_name: Name of the control (denormalized for queries)
check_stage: "pre" (before execution) or "post" (after execution)
applies_to: "llm_call" or "tool_call"
applies_to: "llm_call", "tool_call", or a custom step type with a "_call" suffix
action: The action taken (deny, steer, observe)
matched: Whether the control evaluator matched
confidence: Confidence score from the evaluator (0.0-1.0)
Expand Down Expand Up @@ -90,8 +90,10 @@ class ControlExecutionEvent(BaseModel):
check_stage: Literal["pre", "post"] = Field(
..., description="Check stage: 'pre' or 'post'"
)
applies_to: Literal["llm_call", "tool_call"] = Field(
..., description="Type of call: 'llm_call' or 'tool_call'"
applies_to: str = Field(
...,
min_length=1,
description="Type of call or custom step type with a '_call' suffix",
)

# Result
Expand Down Expand Up @@ -278,7 +280,7 @@ class EventQueryRequest(BaseModel):
actions: Filter by actions (deny, steer, observe)
matched: Filter by matched status
check_stages: Filter by check stages (pre, post)
applies_to: Filter by call type (llm_call, tool_call)
applies_to: Filter by call type or custom call type
start_time: Filter events after this time
end_time: Filter events before this time
limit: Maximum number of events to return
Expand Down Expand Up @@ -310,8 +312,8 @@ class EventQueryRequest(BaseModel):
check_stages: list[Literal["pre", "post"]] | None = Field(
default=None, description="Filter by check stages"
)
applies_to: list[Literal["llm_call", "tool_call"]] | None = Field(
default=None, description="Filter by call types"
applies_to: list[str] | None = Field(
default=None, description="Filter by call types or custom call types"
)
start_time: datetime | None = Field(
default=None, description="Filter events after this time"
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
name = "agent-control"
version = "8.7.0"
description = "Agent Control - protect your AI agents with controls"
requires-python = ">=3.12"
requires-python = ">=3.12,<3.15"

[tool.uv.workspace]
members = [
Expand Down
23 changes: 17 additions & 6 deletions sdks/python/src/agent_control/_control_registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
from typing import Any, NotRequired, TypedDict

from ._schema_derivation import derive_schemas
from .validation import ensure_step_type

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -70,7 +71,11 @@ class _RegisteredControl:
# ---------------------------------------------------------------------------


def register(func: Callable[..., Any], policy: str | None = None) -> None:
def register(
func: Callable[..., Any],
policy: str | None = None,
step_type: str | None = None,
) -> None:
"""Register a decorated function's step schema in the registry.

Extracts step metadata from the function and stores it for later retrieval
Expand All @@ -83,11 +88,17 @@ def register(func: Callable[..., Any], policy: str | None = None) -> None:
Args:
func: The original (unwrapped) function being decorated.
policy: Optional policy name (stored as metadata).
step_type: Optional explicit step type. When omitted, tool-like
functions are registered as ``tool`` and other functions as ``llm``.
"""
# Determine step name -- tools typically have .name or .tool_name
tool_name = getattr(func, "name", None) or getattr(func, "tool_name", None)
step_name: str = tool_name if isinstance(tool_name, str) else func.__name__
step_type: str = "tool" if isinstance(tool_name, str) else "llm"
resolved_step_type = (
ensure_step_type(step_type)
if step_type is not None
else ("tool" if isinstance(tool_name, str) else "llm")
)

# Extract description from docstring (first line only)
description: str | None = None
Expand All @@ -100,10 +111,10 @@ def register(func: Callable[..., Any], policy: str | None = None) -> None:
if policy is not None:
metadata["policy"] = policy

key = _step_key(step_type, step_name)
key = _step_key(resolved_step_type, step_name)
registered = _RegisteredControl(
func=func,
step_type=step_type,
step_type=resolved_step_type,
step_name=step_name,
description=description,
metadata=metadata,
Expand All @@ -114,10 +125,10 @@ def register(func: Callable[..., Any], policy: str | None = None) -> None:
logger.debug(
"Overwriting previously registered step '%s' (type=%s)",
step_name,
step_type,
resolved_step_type,
)
_registered_steps[key] = registered
logger.debug("Registered step schema: %s (type=%s)", step_name, step_type)
logger.debug("Registered step schema: %s (type=%s)", step_name, resolved_step_type)


def get_registered_steps() -> list[StepSchemaDict]:
Expand Down
82 changes: 70 additions & 12 deletions sdks/python/src/agent_control/control_decorators.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ async def chat(message: str) -> str:
)
from agent_control.settings import get_settings
from agent_control.tracing import _generate_span_id, get_current_trace_id, get_trace_and_span_ids
from agent_control.validation import ensure_step_type

logger = get_logger(__name__)

Expand Down Expand Up @@ -96,6 +97,7 @@ class ControlContext:
span_id: str
start_time: float
step_name: str | None = None
step_type: str | None = None

# Stats (mutually exclusive: errors vs matches vs non_matches)
total_executions: int = 0
Expand Down Expand Up @@ -126,13 +128,23 @@ def log_end(self) -> None:
def pre_payload(self) -> dict[str, Any]:
"""Build payload for pre-execution check (supports tool call detection)."""
return _create_evaluation_payload(
self.func, self.args, self.kwargs, output=None, step_name=self.step_name
self.func,
self.args,
self.kwargs,
output=None,
step_name=self.step_name,
explicit_step_type=self.step_type,
)

def post_payload(self, output: Any) -> dict[str, Any]:
"""Build payload for post-execution check (supports tool call detection)."""
return _create_evaluation_payload(
self.func, self.args, self.kwargs, output=output, step_name=self.step_name
self.func,
self.args,
self.kwargs,
output=output,
step_name=self.step_name,
explicit_step_type=self.step_type,
)

def process_result(self, result: dict[str, Any], check_stage: str) -> None:
Expand Down Expand Up @@ -478,7 +490,8 @@ def _create_evaluation_payload(
args: tuple,
kwargs: dict,
output: Any = None,
step_name: str | None = None
step_name: str | None = None,
explicit_step_type: str | None = None,
) -> dict[str, Any]:
"""
Create evaluation payload for server, detecting if it's a tool step or LLM step.
Expand All @@ -491,6 +504,7 @@ def _create_evaluation_payload(
kwargs: Function keyword arguments
output: Function output (None for pre-execution)
step_name: Optional explicit step name to override auto-detection
explicit_step_type: Optional explicit step type to override inference
"""
sig = inspect.signature(func)
bound = sig.bind(*args, **kwargs)
Expand All @@ -505,21 +519,40 @@ def _create_evaluation_payload(
getattr(func, "name", None) is not None
or getattr(func, "tool_name", None) is not None
)
step_type = "tool" if is_tool else "llm"
inferred_type = "tool" if is_tool else "llm"
else:
# Auto-detect: Check if function has tool_name from @tool decorator
tool_name = getattr(func, "name", None) or getattr(func, "tool_name", None)
if tool_name:
determined_name = tool_name
step_type = "tool"
inferred_type = "tool"
else:
determined_name = func.__name__
step_type = "llm"
inferred_type = "llm"

step_type = (
ensure_step_type(explicit_step_type)
if explicit_step_type is not None
else inferred_type
)

if step_type not in ("llm", "tool"):
# Custom step types use the complete bound argument mapping. Unlike an
# LLM step, there is no generic convention for selecting one argument
# as the prompt/input for a retriever, trace, session, etc.
return {
"type": step_type,
"name": determined_name,
"input": dict(bound.arguments),
"output": output if isinstance(output, (str, int, float, bool, dict, list)) else (
None if output is None else str(output)
),
}

if step_type == "tool":
# This is a tool step
return {
"type": "tool",
"type": step_type,
"name": determined_name,
"input": dict(bound.arguments),
"output": output if isinstance(output, (str, int, float, bool, dict, list)) else (
Expand All @@ -530,7 +563,7 @@ def _create_evaluation_payload(
# This is an LLM step
input_data = _extract_input_from_args(func, args, kwargs)
return {
"type": "llm",
"type": step_type,
"name": determined_name,
"input": input_data,
"output": output if isinstance(output, (str, int, float, bool, dict, list)) else (
Expand Down Expand Up @@ -699,6 +732,7 @@ async def _execute_with_control(
kwargs: dict,
is_async: bool,
step_name: str | None = None,
step_type: str | None = None,
) -> Any:
"""
Core control execution logic for both async and sync functions.
Expand All @@ -717,6 +751,7 @@ async def _execute_with_control(
kwargs: Keyword arguments for the function
is_async: Whether the wrapped function is async
step_name: Optional explicit step name for control matching
step_type: Optional explicit step type for control matching

Returns:
The result of the wrapped function
Expand Down Expand Up @@ -751,6 +786,7 @@ async def _execute_with_control(
span_id=span_id,
start_time=time.perf_counter(),
step_name=step_name,
step_type=step_type,
)
ctx.log_start()

Expand All @@ -772,7 +808,11 @@ async def _execute_with_control(
ctx.log_end()


def control(policy: str | None = None, step_name: str | None = None) -> Callable[[F], F]:
def control(
policy: str | None = None,
step_name: str | None = None,
step_type: str | None = None,
) -> Callable[[F], F]:
"""
Decorator to apply server-defined controls at this code location.

Expand All @@ -784,6 +824,9 @@ def control(policy: str | None = None, step_name: str | None = None) -> Callable
clarity in code when multiple policies exist.
step_name: Optional custom name for this step. If not provided, uses
the function name.
step_type: Optional custom type for this step. If not provided, tool-like
functions are classified as ``tool`` and other functions as
``llm``.

Returns:
Decorated function
Expand Down Expand Up @@ -835,6 +878,9 @@ async def handle_user_input(user_message: str) -> str:
POST /api/v1/agents/{agent_name}/policies/{policy_id}
POST /api/v1/agents/{agent_name}/controls/{control_id}
"""
if step_type is not None:
step_type = ensure_step_type(step_type)

# The policy parameter is for documentation only - the server evaluates
# controls associated with the agent via policy and direct links.
_ = policy
Expand All @@ -843,12 +889,17 @@ def decorator(func: F) -> F:
# Register this function's step schema for auto-discovery by init()
from agent_control._control_registry import register

register(func, policy)
register(func, policy=policy, step_type=step_type)

@functools.wraps(func)
async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
return await _execute_with_control(
func, args, kwargs, is_async=True, step_name=step_name
func,
args,
kwargs,
is_async=True,
step_name=step_name,
step_type=step_type,
)

# Copy over ALL attributes from the original function (important for LangChain tools)
Expand All @@ -862,7 +913,14 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
@functools.wraps(func)
def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
return asyncio.run(
_execute_with_control(func, args, kwargs, is_async=False, step_name=step_name)
_execute_with_control(
func,
args,
kwargs,
is_async=False,
step_name=step_name,
step_type=step_type,
)
)

if inspect.iscoroutinefunction(func):
Expand Down
6 changes: 4 additions & 2 deletions sdks/python/src/agent_control/evaluation.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
from .evaluation_events import build_control_execution_events, enqueue_observability_events
from .observability import is_observability_enabled
from .tracing import get_trace_and_span_ids
from .validation import ensure_agent_name
from .validation import ensure_agent_name, ensure_step_type

_RuntimePostEvaluation = Callable[..., Awaitable[httpx.Response]]

Expand Down Expand Up @@ -523,7 +523,7 @@ async def evaluate_controls(
context: dict[str, Any] | None = None,
tools: list[dict[str, JSONValue]] | None = None,
ground_truth: JSONValue | None = None,
step_type: Literal["tool", "llm"] = "llm",
step_type: str = "llm",
stage: Literal["pre", "post"] = "pre",
agent_name: str,
target_type: str | None = None,
Expand All @@ -541,6 +541,8 @@ async def evaluate_controls(
the cached controls were fetched for the session target and would
otherwise drive stale local-first evaluation.
"""
step_type = ensure_step_type(step_type)

if state.server_url is None:
raise RuntimeError("Server URL not configured. Call agent_control.init() first.")

Expand Down
Loading
Loading