Appendix E: HR Leave Agent — Code Reference

Version: 1.0 Last updated: February 2026

Navigation: ← Appendix D | Appendix Index | Appendix F →

This appendix provides key code excerpts from the HR Leave Agent implementation. Use these as templates for your own agent development.

📦 Full Implementation: The complete runnable code is available at:
github.com/sunilsivadas/agentic-ai-handbook-code

📚 Related sections: - State machine scaffold → Chapter 4, §4.2.2 - Tool patterns → Chapter 9, §9.2 - Evaluation harness → Chapter 7, §7.2 - Prompts and rubrics → Appendix C - Synthetic datasets → Appendix D

---

E.1 Repository Structure

agentic-ai-handbook-code/
├── README.md
├── requirements.txt
├── pyproject.toml
│
├── hr_leave_agent/
│   ├── __init__.py
│   ├── agent.py          # State machine scaffold (~600 lines)
│   ├── tools.py          # Mock HR system tools (~740 lines)
│   ├── prompts.py        # All prompts by state (~440 lines)
│   └── run.py            # Interactive demo (~270 lines)
│
├── eval/
│   ├── __init__.py
│   ├── evaluate.py       # Evaluation harness (~680 lines)
│   ├── judges/
│   │   ├── eligibility_judge.py
│   │   ├── safety_judge.py
│   │   ├── citation_judge.py
│   │   ├── quality_judge.py
│   │   └── composite_judge.py
│   └── datasets/
│       └── hr_leave/
│           ├── manifest.yaml
│           ├── golden/
│           ├── adversarial/
│           └── regression/
│
├── config/
│   ├── config.yaml
│   └── eval_config.yaml
│
└── docs/
    ├── prompts-rubrics.md
    └── synthetic-datasets.md

---

E.2 State Machine Scaffold

The core pattern: states with specific responsibilities, allowed tools, and transition rules.

E.2.1 State Definitions

from enum import Enum
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

class State(Enum):
    """Agent states in the scaffold."""
    INTAKE = "intake"      # Parse user intent
    RETRIEVE = "retrieve"  # Gather context via tools
    ASSESS = "assess"      # Evaluate eligibility
    RESPOND = "respond"    # Generate user response
    PREVIEW = "preview"    # Show submission preview
    SUBMIT = "submit"      # Execute write operation
    END = "end"            # Terminal state


@dataclass
class AgentContext:
    """Holds all context accumulated during agent execution."""
    employee_id: str
    conversation_history: List[Dict] = field(default_factory=list)

    # Parsed from intake
    parsed_intent: Optional[Dict] = None

    # Retrieved data
    employee_profile: Optional[Dict] = None
    leave_balances: Optional[Dict] = None
    policy_excerpts: List[Dict] = field(default_factory=list)
    calendar_data: Optional[Dict] = None

    # Assessment
    assessment: Optional[Dict] = None

    # Submission
    request_preview: Optional[Dict] = None
    submitted_request: Optional[Dict] = None
    idempotency_key: Optional[str] = None

    # Tracking
    current_state: State = State.INTAKE
    step_count: int = 0
    tool_calls: List[Dict] = field(default_factory=list)

E.2.2 State Machine Loop

class HRLeaveAgent:
    """
    HR Leave Agent using state machine scaffold.

    State flow:
        INTAKE → RETRIEVE → ASSESS → RESPOND → (END or PREVIEW)
                                         ↓
                                      PREVIEW → SUBMIT → RESPOND → END
    """

    MAX_STEPS = 10

    def __init__(self, model: str = "claude-sonnet-4-20250514"):
        self.client = anthropic.Anthropic()
        self.model = model

    async def run(self, user_message: str, employee_id: str) -> Dict[str, Any]:
        """Run the agent for a single user message."""
        context = AgentContext(employee_id=employee_id)
        context.conversation_history.append({
            "role": "user",
            "content": user_message
        })
        return await self._run_state_machine(context)

    async def _run_state_machine(self, context: AgentContext) -> Dict[str, Any]:
        """Execute the state machine until terminal state or max steps."""

        while context.current_state != State.END and context.step_count < self.MAX_STEPS:
            context.step_count += 1

            # Execute current state
            next_state, output = await self._execute_state(context)

            # Transition
            context.current_state = next_state

            # If we have a response to return, break
            if output.get("response"):
                context.conversation_history.append({
                    "role": "assistant",
                    "content": output["response"]
                })
                return {
                    "response": output["response"],
                    "context": context,
                    "state": context.current_state.value,
                    "tool_calls": context.tool_calls,
                }

        # Fallback if we hit max steps
        return {
            "response": "I apologize, but I'm having trouble processing your request. "
                       "Please try rephrasing or contact HR directly.",
            "context": context,
            "state": context.current_state.value,
            "error": "max_steps_exceeded" if context.step_count >= self.MAX_STEPS else None,
        }

    async def _execute_state(self, context: AgentContext) -> tuple[State, Dict]:
        """Execute a single state and return next state + output."""

        state = context.current_state

        if state == State.INTAKE:
            return await self._execute_intake(context)
        elif state == State.RETRIEVE:
            return await self._execute_retrieve(context)
        elif state == State.ASSESS:
            return await self._execute_assess(context)
        elif state == State.RESPOND:
            return await self._execute_respond(context)
        elif state == State.PREVIEW:
            return await self._execute_preview(context)
        elif state == State.SUBMIT:
            return await self._execute_submit(context)
        else:
            return State.END, {"response": "Unknown state"}

E.2.3 State-Specific Tool Restrictions

def get_tools_for_state(state: str) -> List[Dict]:
    """Get tool definitions allowed for a given state."""
    state_tools = {
        "intake": [],  # No tools - pure parsing
        "retrieve": [
            "get_employee_profile",
            "check_leave_balance",
            "get_policy_sections",
            "check_calendar"
        ],
        "assess": [],  # No tools - pure reasoning
        "respond": [],  # No tools - pure generation
        "preview": ["prepare_leave_request"],
        "submit": ["submit_leave_request"],
    }

    allowed = state_tools.get(state, [])
    return [t for t in TOOL_DEFINITIONS if t["name"] in allowed]

---

E.3 Tool Patterns

Tools follow strict contracts with typed inputs/outputs and side-effect classification.

E.3.1 Tool Definitions

TOOL_DEFINITIONS = [
    {
        "name": "get_employee_profile",
        "description": "Get employee profile including name, department, grade, and dependents.",
        "input_schema": {
            "type": "object",
            "properties": {
                "employee_id": {
                    "type": "string",
                    "description": "The employee ID to look up"
                }
            },
            "required": ["employee_id"]
        }
    },
    {
        "name": "check_leave_balance",
        "description": "Check leave balance for an employee.",
        "input_schema": {
            "type": "object",
            "properties": {
                "employee_id": {"type": "string"},
                "leave_type": {
                    "type": "string",
                    "enum": ["annual", "sick", "childcare", "maternity", "paternity"]
                }
            },
            "required": ["employee_id"]
        }
    },
    {
        "name": "prepare_leave_request",
        "description": "Prepare a leave request for preview. Does NOT submit.",
        "input_schema": {
            "type": "object",
            "properties": {
                "employee_id": {"type": "string"},
                "leave_type": {"type": "string"},
                "start_date": {"type": "string", "description": "YYYY-MM-DD format"},
                "end_date": {"type": "string", "description": "YYYY-MM-DD format"},
                "reason": {"type": "string"}
            },
            "required": ["employee_id", "leave_type", "start_date", "end_date"]
        }
    },
    {
        "name": "submit_leave_request",
        "description": "Submit a prepared leave request. Requires idempotency_key.",
        "input_schema": {
            "type": "object",
            "properties": {
                "request_id": {"type": "string"},
                "idempotency_key": {
                    "type": "string",
                    "description": "Unique key to prevent duplicate submissions"
                }
            },
            "required": ["request_id", "idempotency_key"]
        }
    }
]

E.3.2 Idempotent Write Operation

def submit_leave_request(
    self,
    request_id: str,
    idempotency_key: str
) -> Dict[str, Any]:
    """
    Submit a prepared leave request.

    Uses idempotency_key to ensure exactly-once semantics:
    - If the same key is used twice, returns the original result
    - Prevents duplicate submissions on retry
    """
    # Check idempotency
    if idempotency_key in self.idempotency_keys:
        existing = self.submitted_requests.get(request_id)
        if existing:
            return {
                "request_id": existing.request_id,
                "status": existing.status,
                "submitted_at": existing.created_at,
                "note": "Duplicate submission detected, returning original result"
            }

    # Mark idempotency key as used
    self.idempotency_keys.add(idempotency_key)

    # Create the submission
    final_request_id = request_id.replace("DRAFT", "2025")
    submitted_at = datetime.now().isoformat()

    request = LeaveRequest(
        request_id=final_request_id,
        status="pending",
        created_at=submitted_at,
        # ... other fields
    )

    self.submitted_requests[final_request_id] = request

    return {
        "request_id": final_request_id,
        "status": "pending",
        "submitted_at": submitted_at,
        "message": "Leave request submitted. Awaiting manager approval."
    }

---

E.4 Evaluation Harness

The harness runs test cases through the agent and evaluates with multiple judges.

E.4.1 Test Case Structure

from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

@dataclass
class TestCase:
    """A single evaluation test case."""
    id: str
    name: str
    category: str
    complexity: str
    tags: List[str]
    input: Dict[str, Any]
    expected: Dict[str, Any]


@dataclass
class JudgeResult:
    """Result from a single judge."""
    judge_name: str
    scores: Dict[str, float]
    passed: bool
    justifications: Dict[str, str] = field(default_factory=dict)
    issues: List[str] = field(default_factory=list)


@dataclass
class EvalResult:
    """Result for a single test case."""
    case_id: str
    case_name: str
    passed: bool
    overall_score: float
    judge_results: Dict[str, JudgeResult]
    agent_response: str
    tool_calls: List[Dict]
    error: Optional[str] = None
    critical_failures: List[str] = field(default_factory=list)

E.4.2 Evaluation Harness

class EvaluationHarness:
    """Runs evaluation suites against an agent."""

    def __init__(self, agent, judges: List[LLMJudge] = None):
        self.agent = agent
        self.judges = judges or [
            EligibilityJudge(),
            SafetyJudge(),
            QualityJudge(),
        ]
        self.rule_checker = RuleBasedChecker()

    def load_cases(self, path: str) -> List[TestCase]:
        """Load test cases from YAML file."""
        with open(path) as f:
            data = yaml.safe_load(f)

        return [
            TestCase(
                id=case["id"],
                name=case["name"],
                category=case["category"],
                complexity=case["complexity"],
                tags=case.get("tags", []),
                input=case["input"],
                expected=case["expected"],
            )
            for case in data.get("cases", [])
        ]

    def run_evaluation(
        self,
        cases: List[TestCase],
        filter_tags: List[str] = None,
        verbose: bool = True
    ) -> EvalReport:
        """Run evaluation on all cases."""

        # Filter if needed
        if filter_tags:
            cases = [c for c in cases if any(t in c.tags for t in filter_tags)]

        results = []
        passed = 0
        failed = 0
        non_negotiable_failures = []

        for case in cases:
            if verbose:
                print(f"Evaluating: {case.id} - {case.name}")

            result = self._evaluate_case(case)
            results.append(result)

            if result.passed:
                passed += 1
            else:
                failed += 1
                if result.critical_failures:
                    non_negotiable_failures.extend([
                        f"{case.id}: {f}" for f in result.critical_failures
                    ])

        return EvalReport(
            timestamp=datetime.now().isoformat(),
            total_cases=len(cases),
            passed=passed,
            failed=failed,
            pass_rate=passed / len(cases) if cases else 0,
            results=results,
            non_negotiable_failures=non_negotiable_failures,
            metrics=self._aggregate_metrics(results),
        )

    def _evaluate_case(self, case: TestCase) -> EvalResult:
        """Evaluate a single test case."""

        # Run agent
        try:
            result = self.agent.run(
                case.input["user_query"],
                case.input["employee_id"]
            )
            agent_response = result.get("response", "")
            tool_calls = result.get("tool_calls", [])
        except Exception as e:
            return EvalResult(
                case_id=case.id,
                case_name=case.name,
                passed=False,
                overall_score=0,
                judge_results={},
                agent_response="",
                tool_calls=[],
                error=str(e),
                critical_failures=[f"Agent error: {str(e)}"]
            )

        # Run rule-based checks (fast)
        critical_failures = []

        if "must_mention" in case.expected:
            missing = self.rule_checker.check_must_mention(
                agent_response, case.expected["must_mention"]
            )
            if missing:
                critical_failures.append(f"Missing required terms: {missing}")

        if "must_not_mention" in case.expected:
            violations = self.rule_checker.check_must_not_mention(
                agent_response, case.expected["must_not_mention"]
            )
            if violations:
                critical_failures.append(f"Contains forbidden terms: {violations}")

        # Run LLM judges
        judge_results = {}
        for judge in self.judges:
            try:
                judge_result = judge.evaluate(case, agent_response, tool_calls)
                judge_results[judge.name] = judge_result

                if not judge_result.passed and judge.name in ["eligibility", "safety"]:
                    critical_failures.extend(judge_result.issues)
            except Exception as e:
                judge_results[judge.name] = JudgeResult(
                    judge_name=judge.name,
                    scores={},
                    passed=False,
                    issues=[f"Judge error: {str(e)}"]
                )

        # Calculate overall score and pass/fail
        overall_score = self._calculate_score(judge_results)
        passed = (
            len(critical_failures) == 0 and
            all(jr.passed for name, jr in judge_results.items() if name != "quality")
        )

        return EvalResult(
            case_id=case.id,
            case_name=case.name,
            passed=passed,
            overall_score=overall_score,
            judge_results=judge_results,
            agent_response=agent_response,
            tool_calls=tool_calls,
            critical_failures=critical_failures,
        )

---

E.5 Interactive Demo

The companion repository includes an interactive demo. Run it with:

python -m hr_leave_agent.run

The demo lets you role-play as one of four sample employees (varying balances, dependents, and edge cases) and test queries against the agent.

---

E.6 Running the Code

Installation

# Clone the repository
git clone https://github.com/sunilsivadas/agentic-ai-handbook-code.git
cd agentic-ai-handbook-code

# Create virtual environment
python -m venv venv
source venv/bin/activate  # or `venv\Scripts\activate` on Windows

# Install dependencies
pip install -r requirements.txt

# Set API key
export ANTHROPIC_API_KEY='your-key-here'

Interactive Demo

# Run interactive demo
python -m hr_leave_agent.run

# Or run a single query
python -m hr_leave_agent.run --query "Can I take 3 days off next week?" --employee EMP-2024-001

Run Evaluation

# Run against sample cases
python -m eval.evaluate --cases eval/datasets/sample_cases.yaml --output eval_report.json

# Run with tag filter
python -m eval.evaluate --cases eval/datasets/sample_cases.yaml --tags childcare --output childcare_report.json

# Run adversarial tests
python -m eval.evaluate --cases eval/datasets/hr_leave/adversarial --output adversarial_report.json

---

E.7 Requirements

# requirements.txt
anthropic>=0.18.0
pyyaml>=6.0
python-dateutil>=2.8.2

---

📦 Full Implementation: Clone the repository to explore the complete code, run the demo, and adapt for your own use cases. Design rationale (why state machine over ReAct, why idempotency keys, why separate judges) is discussed in Chapter 4 §4.2, Chapter 9 §9.2, and Chapter 7 §7.1 respectively.

---

Navigation: ← Appendix D | Appendix Index | Appendix F →