hook-event-architecture

Hook Event Architecture

Design hook-based event systems for capturing and broadcasting agent activities in AI Developer Workflows.

MANDATORY: docs-management Delegation

Documentation Verification: This skill references Claude Code hook events and configuration that may change between releases. Before implementing, invoke hook-management skill to verify current event types and hook configuration patterns.

Verification Checkpoint:

Did I invoke hook-management for current event types?
Did official documentation confirm the event schema?
Is my implementation based on official docs?

When to Use

Implementing real-time event broadcasting
Building observability infrastructure
Creating swimlane visualizations
Logging agent activities to database
Generating AI-powered event summaries

Prerequisites

Understanding of Claude Code hooks (@hook-event-patterns.md)
Familiarity with WebSocket patterns (@websocket-architecture.md)
Access to Claude Agent SDK for full implementation

SDK Requirement

Implementation Note: Full hook event architecture requires Claude Agent SDK with custom tooling. This skill provides design patterns and specifications.

Event Types

Documentation Verification: The event types listed here are Claude Code internal types that may change between releases. For authoritative current event types, query the hook-management skill which delegates to docs-management for official documentation.

ADW systems capture these event types:

Event Type Icon Source Payload

PreToolUse

🪝 Hook Tool name, inputs, session

PostToolUse

🪝 Hook Tool name, outputs, duration

TextBlock

💬 Agent Response text, tokens

ToolUseBlock

🛠️ Agent Tool invocation record

ThinkingBlock

🧠 Agent Extended thinking content

StepStart

⚙️ System Step name, inputs

StepEnd

⚙️ System Step name, outputs, duration

Architecture Design Process

Step 1: Define Event Schema

Create Pydantic models for events:

class ADWEvent(BaseModel): type: str # Event type from table adw_id: str # 8-char correlation ID step: str # Current step name timestamp: datetime payload: dict # Type-specific data summary: str | None # AI-generated summary

Step 2: Configure Hook Triggers

Set up Claude Code hooks:

{ "hooks": { "PreToolUse": [{ "matcher": ".", "command": "python hooks/pre_tool.py" }], "PostToolUse": [{ "matcher": ".", "command": "python hooks/post_tool.py" }] } }

Step 3: Design Event Pipeline

Agent Execution │ ├── PreToolUse ──► Hook Script ──┬── Log to DB │ ├── Summarize (Haiku) ▼ └── Broadcast (WebSocket) Tool Execution │ ├── PostToolUse ──► Hook Script ──┬── Log to DB │ ├── Summarize (Haiku) ▼ └── Broadcast (WebSocket) Continue...

Step 4: Implement Summarization

AI-generated event summaries using Haiku:

async def summarize_event(event: ADWEvent) -> str: prompt = f"""Summarize in 15 words or less: Event: {event.type} Tool: {event.payload.get('tool_name', 'N/A')} Data: {str(event.payload)[:500]} """ return await claude.complete(prompt, model="haiku")

Step 5: Design Broadcast Pattern

Event distribution to clients:

class EventBroadcaster: def init(self, ws_manager, db_client): self.ws = ws_manager self.db = db_client

async def broadcast(self, event: ADWEvent):
    # Log to database first
    await self.db.log_event(event)

    # Broadcast to WebSocket clients
    await self.ws.broadcast(event.dict())

Hook Script Templates

PreToolUse Hook

#!/usr/bin/env python import sys, json, asyncio from adw_modules import broadcast, summarize

async def main(): data = json.load(sys.stdin)

event = {
    "type": "PreToolUse",
    "adw_id": data.get("adw_id"),
    "step": data.get("step"),
    "payload": {
        "tool_name": data["tool_name"],
        "tool_input": data["tool_input"]
    }
}

event["summary"] = await summarize(event)
await broadcast(event)

if name == "main": asyncio.run(main())

PostToolUse Hook

#!/usr/bin/env python import sys, json, asyncio from adw_modules import broadcast, summarize

async def main(): data = json.load(sys.stdin)

event = {
    "type": "PostToolUse",
    "adw_id": data.get("adw_id"),
    "step": data.get("step"),
    "payload": {
        "tool_name": data["tool_name"],
        "tool_output": data.get("tool_output", "")[:1000],
        "duration_ms": data.get("duration_ms", 0)
    }
}

event["summary"] = await summarize(event)
await broadcast(event)

if name == "main": asyncio.run(main())

Output Format

When designing hook event architecture: