google / google/adk-python

[Feature Request]: Add framework-neutral agent-hooks contract to govern ADK agents

Open
#6,602 1 comment 0 reactions 1 assignee Claimed by @sanketpatil06 View on GitHub
agent engine needs review
Dominant language
Python
Stars
21.5k
Forks
4k
Avg merge
1d 14h
Merged PRs (30d)
37

Description

** Please make sure you read the contribution guide and file the issues in the right place. **
[Contribution guide.](https://google.github.io/adk-docs/contributing-guide/)

## 🔴 Required Information

### Is your feature request related to a specific problem?
ADK exposes rich lifecycle seams through `BasePlugin`, but there is no
standard, framework-neutral way to plug governance and policy enforcement
(policy engines, content filters, PII redaction, egress guards, rate
limiters) into those seams. Today each team hand-writes bespoke callback
logic per agent. That is inconsistent, easy to accidentally "fail open,"
hard to audit, and not portable across agent frameworks. There is no
uniform allow / deny / transform decision contract, no fail-closed
guarantee, and no built-in, auditable record of *why* an action was
allowed, blocked, or rewritten.

### Describe the Solution You'd Like
Add `AgentHooksPlugin`: a `BasePlugin` that makes ADK a conformant *host*
for [agent-hooks](https://github.com/responsibleai/agent-hooks), a
framework-neutral control contract. You register one or more interceptors
once, and every governed ADK lifecycle point delegates its decision to the
agent-hooks emitter, which runs the interceptors, returns a `Verdict`
(`allow` / `deny` / `transform`), and records each decision as an auditable
`InterceptionRecord`.

API changes (all additive):
- New public export `google.adk.plugins.AgentHooksPlugin`.
- New optional extra: `pip install "google-adk[agent-hooks]"` (backed by
`agent-hooks-sdk`, a compiled-core optional dependency). The import is
deferred to plugin construction, so importing `google.adk.plugins` never
requires it.
- Interceptor contract: any object with `intercept(AgentContext) -> Verdict`
(sync or async). No ADK-specific base class required — the same
interceptor runs on other agent-hooks hosts (crewAI, ...).

Enforcement is **fail-closed**: `deny` blocks the guarded action,
`transform` rewrites the guarded value at `$target`, and any engine error,
malformed verdict, or interceptor timeout becomes a fail-closed deny — it
never fails open. `mode="evaluate_only"` records verdicts without acting.

### Impact on your work
This gives us one consistent, auditable governance layer across ADK agents
instead of per-agent callback code, and lets teams
enforce the *same* interceptor across ADK and other frameworks. It is
foundational for governed/regulated deployments where every tool call,
model call, input, and output must be policy-checked and logged.

### Willingness to contribute
Yes.

---

## 🟡 Recommended Information

### Describe Alternatives You've Considered
- **Framework-specific middleware.** Locks policy to one framework;
agent-hooks is framework-neutral, so the same interceptor is portable.
- **Hand-written `BasePlugin` callbacks per project.** Works, but is
reimplemented everywhere, has no shared verdict model or audit trail, and
is easy to accidentally fail open.
- **Provider/model-side guardrails only.** They don't uniformly cover tool
calls, tool results, user input, final output, and run lifecycle.

### Proposed API / Implementation
```python
from google.adk.apps.app import App
from google.adk.plugins import AgentHooksPlugin
from agent_hooks import AgentContext, Decision, Verdict

class ToolGovernance:
name = "tool_governance"

def intercept(self, ctx: AgentContext) -> Verdict:
if ctx["interception_point"] == "pre_tool_call":
if ctx["tool_call"]["name"] == "delete_account":
return Verdict.deny(reason="tool_denied",
message="disabled by policy")
return Verdict(decision=Decision.ALLOW)

records = []
plugin = AgentHooksPlugin(
interceptors=[ToolGovernance()],
mode="enforce", # or "evaluate_only" to observe without acting
record_sink=records.append, # one auditable InterceptionRecord per decision
)
app = App(name="demo", root_agent=root_agent, plugins=[plugin])
```
Constructor (defaults shown): `AgentHooksPlugin(interceptors, name="agent_hooks",
mode="enforce", timeout=5.0, composition=None, identity_provider="jcs-sha256",
record_sink=None, max_records=1000)`.

### Additional Context
Interception-point mapping (ADK callback → agent-hooks point):

| ADK plugin callback | agent-hooks point |
| ----------------------------- | ----------------- |
| `before_run_callback` | `agent_startup` |
| `on_user_message_callback` | `input` |
| `before_model_callback` | `pre_model_call` |
| `after_model_callback` | `post_model_call` |
| `before_tool_callback` | `pre_tool_call` |
| `after_tool_callback` | `post_tool_call` |
| `on_event_callback` (final) | `output` |
| `after_run_callback` | `agent_shutdown` |

Notes:
- `pre_model_call` supports `allow`/`deny` only; a `transform` there is
treated as a fail-closed deny, because rebuilding a provider-native
request from wire messages is not round-trip safe.
- **Trust model:** agent-hooks is a *cooperative* control contract, **not**
a security boundary. Interceptors run in-process with full data access and
the interception points do not guarantee complete mediation.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.