Custom Server Plugins

Server plugins run on the AgentGuard control server. They are useful for centralized policy decisions, cross-step detection, and checks that need recent session history.

Server plugin files should be placed under the phase folder that matches the event type:

src/server/backend/runtime/plugins/llm_before/
src/server/backend/runtime/plugins/llm_after/
src/server/backend/runtime/plugins/tool_before/
src/server/backend/runtime/plugins/tool_after/

Input

A server plugin implements this method:

def check(
    self,
    event: RuntimeEvent,
    context: RuntimeContext,
    trajectory_window: list[RuntimeEvent] | None = None,
) -> CheckResult:
    ...

The server plugin manager calls check() only when the current event phase matches the configured phase and event_types allows the event.

event: RuntimeEvent

event is the normalized runtime event that the plugin should inspect:

RuntimeEvent(
    event_id: str,
    event_type: EventType,
    timestamp: float,
    context: RuntimeContext,
    payload: LLMInput | LLMOutput | ToolInvoke | ToolResult,
    risk_signals: list[str] = [],
    metadata: dict[str, Any] = {},
)

• event_id: unique identifier for the event.
• event_type: current event type. Supported values are LLM_INPUT, LLM_OUTPUT, TOOL_INVOKE, and TOOL_RESULT.
• timestamp: event creation time.
• context: the same runtime context passed as the second argument.
• payload: one of the four typed payload classes: LLMInput, LLMOutput, ToolInvoke, or ToolResult.
• risk_signals: risk labels already attached by client plugins, preprocessors, or earlier server plugins.
• metadata: adapter-specific or debug metadata.

Common payload shapes:

# LLM_INPUT
LLMInput(messages=[{"role": "user", "content": "..."}])

# LLM_OUTPUT
LLMOutput(output="...", thought=None, final_output=None)

# TOOL_INVOKE
ToolInvoke(
    tool_name="send_email",
    arguments={"to": "...", "body": "..."},
    capabilities=["external_send"],
)

# TOOL_RESULT
ToolResult(tool_name="read_file", result="...")

For LLMOutput, the three fields have different purposes:

• payload.output: canonical text for backward compatibility and general-purpose policy checks.
• payload.thought: optional hidden reasoning text when the adapter can separate it.
• payload.final_output: optional user-visible final answer.

In most plugins, start with payload.output. Only read payload.thought or payload.final_output when your logic explicitly cares about that distinction.

context: RuntimeContext

context describes the current session and agent identity:

RuntimeContext(
    session_id: str,
    user_id: str | None = None,
    agent_id: str | None = None,
    task_id: str | None = None,
    policy: str | None = None,
    policy_version: str | None = None,
    environment: str | None = None,
    metadata: dict[str, Any] = {},
)

• session_id: required session identifier.
• user_id: optional end-user identity.
• agent_id: optional agent instance or service identity.
• task_id: optional workflow or task identifier.
• policy: optional policy name, source, or mode.
• policy_version: optional policy version or snapshot identifier.
• environment: optional runtime environment such as dev, staging, or prod.
• metadata: free-form extra context.

trajectory_window: list[RuntimeEvent] | None

trajectory_window is only available to server plugins.

• It contains recent events from the same session.
• Each item is a full RuntimeEvent.
• It can include client-side cached plugin decisions that were synchronized to the server.
• Use it for multi-step checks, such as detecting sensitive data read in one tool result and then sent through a later outbound tool call.

Always handle None defensively:

trajectory_window = trajectory_window or []

Configuration Input

Server plugin specs are read from the server list in config/plugins.json or the runtime plugin config:

{
  "phases": {
    "tool_before": {
      "client": [],
      "server": [
        {
          "name": "my_server_plugin",
          "env": {}
        }
      ]
    }
  }
}

• name: registered plugin name.
• class or plugin: optional import-path alternatives to name.
• The current server runtime resolves plugin classes by name or import path.
• kwargs values are passed to the plugin constructor. env maps constructor attribute names to environment-variable references such as $MY_PLUGIN_API_KEY; references are resolved in the server process.

Output

check() must return CheckResult:

@dataclass
class CheckResult:
    decision_candidate: GuardDecision | None = None
    risk_signals: list[str] = field(default_factory=list)
    is_final: bool = True
    metadata: dict[str, Any] = field(default_factory=dict)

• decision_candidate: optional GuardDecision recommendation. Use it when the plugin wants to propose ALLOW (allow immediately), DENY (block execution), LOG_ONLY (annotate it in audit logs but still allow it), or HUMAN_CHECK (keep the action pending until a user decides on the server whether to allow it).
• risk_signals: risk labels detected by this plugin. The manager deduplicates them and writes them back to event.risk_signals.
• is_final: whether decision_candidate should be treated as the authoritative server-side decision. It defaults to True. If True, the runtime can use this decision directly.
• metadata: structured debug or detection details. The manager merges plugin metadata into the final plugin result.

Return CheckResult.empty() when the plugin has no finding.

Example

from backend.runtime.plugins.base import BasePlugin, CheckResult
from backend.runtime.plugins.registry import register
from shared.schemas.context import RuntimeContext
from shared.schemas.decisions import GuardDecision
from shared.schemas.events import EventType, RuntimeEvent


@register(
    name="my_server_plugin",
    description="Detect multi-step exfiltration on the server side.",
)
class MyServerPlugin(BasePlugin):
    event_types = [EventType.TOOL_INVOKE]

    def check(
        self,
        event: RuntimeEvent,
        context: RuntimeContext,
        trajectory_window: list[RuntimeEvent] | None = None,
    ) -> CheckResult:
        trajectory_window = trajectory_window or []
        tool_name = event.payload.tool_name

        saw_sensitive_read = any(
            item.event_type == EventType.TOOL_RESULT
            and "secret_detected" in (item.risk_signals or [])
            for item in trajectory_window
        )

        if saw_sensitive_read and tool_name == "send_email":
            return CheckResult(
                decision_candidate=GuardDecision.deny(
                    "Sensitive data cannot be sent by email.",
                    policy_id="server:block_exfiltration",
                    risk_signals=["cross_step_exfiltration"],
                ),
                risk_signals=["cross_step_exfiltration"],
                is_final=True,
                metadata={"trajectory_events": len(trajectory_window)},
            )

        return CheckResult.empty()

Configuration:

{
  "phases": {
    "tool_before": {
      "client": [],
      "server": ["my_server_plugin"]
    }
  }
}