agenthooks
Named hook points for production AI agents.
Customers extend your agent — no fork, no source changes, nothing breaks.
$ pip install agenthooks-py
Named hook points for production AI agents.
Customers extend your agent — no fork, no source changes, nothing breaks.
You build a production AI agent. A customer deploys it. They need approval gates, compliance enrichment, rate limiting, custom audit logging. Today they fork your code. You lose control of the release cycle.
# Without agenthooks — customer forks, patches, diverges forever # With agenthooks — customer registers logic, agent stays canonical class MyAgent(HookAgent): before_call = hookpoint("before_call") # ← agent author declares this once async def run(self, query): ctx = HookContext.new(session_id="s1", tenant_id="acme", query=query) async with self.before_call.run(ctx) as ctx: return await llm.call(ctx.query) # hooks have already enriched ctx # Customer code — entirely separate file, separate team, separate deploy registry = HookRegistry() @registry.implement("before_call") async def inject_context(ctx): return ctx.enrich("region", "EU").enrich("tier", "enterprise") agent = MyAgent(registries=[registry])
Declare extension points anywhere in your agent pipeline. Sequential or parallel execution. Per-hook timeouts and fallback behaviour.
Filter conditions are enforced by the executor. A hook registered for tenant A never fires for tenant B. Multiple customers coexist on one agent.
Every hook execution emits a span parented to your active trace, 5 metric instruments, and trace-correlated structured logs. Zero config required.
Every execution — ok, timeout, error, blocked — writes a JSONL entry. Cannot be disabled. Redacted fields surface as [REDACTED].
Sealed context fields prevent tenant impersonation. Prompt injection scanning on modified queries. Redaction API for credentials and PII.
@inject, @block_if, @redact, @rate_limit, @require_tenant, @retry — compose freely, zero boilerplate.
Configure the OTel SDK once. Hook spans appear as children under your application trace in Jaeger, Tempo, Datadog APM, Honeycomb — any OTLP backend.
from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter provider = TracerProvider() provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter())) trace.set_tracer_provider(provider) # agenthooks picks this up automatically. # Every hook execution becomes a child span with hook.name, hook.impl, # hook.tenant_id, hook.status, hook.duration_ms attributes.
| Metric | Type | Dimensions |
|---|---|---|
| agenthooks.hook.executions | Counter | hook.name · hook.impl · hook.status |
| agenthooks.hook.duration_ms | Histogram | hook.name · hook.impl |
| agenthooks.hook.errors | Counter | hook.name · hook.impl |
| agenthooks.hook.timeouts | Counter | hook.name |
| agenthooks.hook.blocked | Counter | hook.name · hook.impl |
Three-layer design, data model, execution flow, registry filter evaluation, and observability wiring.
Threat model, sealed fields, injection scanning, tenant isolation, redaction, audit invariant.
Step-by-step trace of a hook execution: span creation, metric recording, audit write, all failure paths.
5 runnable examples: basic hooks, customer extensibility, multi-tenant isolation, resilience, OTel wiring.