Introduction
This seventh agent design pattern focuses on a critical process in every organization—Employee Onboarding. The Employee Onboarding Agent assists HR departments by automating the document collection, policy acknowledgments, equipment provisioning, and system access setup needed when an employee joins the company. The agent validates the data, makes necessary approvals, and only claims success when it executes actions and produces receipts for each completed task. The agent reduces human error, increases efficiency, and ensures compliance in an otherwise tedious and error-prone process.
The Use Case
A new employee is joining a company, and the onboarding process includes multiple tasks such as collecting personal details, confirming policy acknowledgment, assigning workstations, provisioning access to tools, and assigning mentors. The agent is responsible for collecting personal data, ensuring that required documents are signed, making sure all approvals are in place, and finally provisioning the employee’s access to systems and equipment. Each step is tracked, and no action is finalized until the system can confirm a receipt.
Prompt Contract (agent interface)
# file: contracts/onboarding_v1.yaml
role: "EmployeeOnboardingAgent"
scope: >
Collect personal details, confirm signed documents, assign equipment, and provision system access.
Ask once for missing fields (employee_id, personal_info, policy_acknowledgment, equipment_requested, access_requested).
Propose tool calls; never assert success without a receipt.
output:
type: object
required: [summary, decision, employee_data, next_steps, tool_proposals]
properties:
summary: {type: string, maxWords: 80}
decision: {type: string, enum: ["approve","reject","need_approval","need_more_info"]}
employee_data:
type: object
required: [employee_id, name, department, equipment_assigned, system_access]
properties:
employee_id: {type: string}
name: {type: string}
department: {type: string}
equipment_assigned: {type: array, items: {type: string}, maxItems: 3}
system_access: {type: array, items: {type: string}, maxItems: 5}
citations: {type: array, items: {type: string}}
next_steps: {type: array, items: {type: string}, maxItems: 6}
tool_proposals:
type: array
items:
type: object
required: [name, args, preconditions, idempotency_key]
properties:
name: {type: string, enum: [CollectPersonalData, VerifyPolicyAcknowledgment, AssignEquipment, ProvisionSystemAccess]}
args: {type: object}
preconditions: {type: string}
idempotency_key: {type: string}
policy_id: "onboarding_policy.v3"
citation_rule: "1–2 minimal-span claim_ids per factual sentence"
decoding:
narrative: {top_p: 0.92, temperature: 0.72, stop: ["\n\n## "]}
bullets: {top_p: 0.82, temperature: 0.45}
Example claims (context given to the model)
[
{"claim_id":"policy:equipment:role_based","text":"Laptops must be assigned based on department (e.g., Finance gets MacBook, Sales gets Windows laptop).",
"effective_date":"2025-01-01","source_id":"doc:onboarding_policy_v3","span":"role-based equipment allocation"},
{"claim_id":"policy:access:tools_required","text":"All employees must have access to company email, HR software, and Slack on day one.",
"effective_date":"2025-01-01","source_id":"doc:onboarding_policy_v3","span":"company email, HR software, and Slack"},
{"claim_id":"policy:documentation:mandatory","text":"The new employee must sign the confidentiality agreement and the data protection policy.",
"effective_date":"2025-01-01","source_id":"doc:onboarding_policy_v3","span":"sign confidentiality agreement, data protection policy"}
]
Tool Interfaces (typed, with receipts)
# tools.py
from pydantic import BaseModel
from typing import List, Optional, Dict
from datetime import date
class CollectPersonalDataArgs(BaseModel):
employee_id: str
name: str
department: str
email: str
phone_number: Optional[str] = None
class VerifyPolicyAcknowledgmentArgs(BaseModel):
employee_id: str
policies_acknowledged: List[str]
class AssignEquipmentArgs(BaseModel):
employee_id: str
equipment: List[str] # e.g., ["laptop", "monitor", "keyboard"]
class ProvisionSystemAccessArgs(BaseModel):
employee_id: str
access_list: List[str] # e.g., ["email", "HR system", "Slack"]
class ToolReceipt(BaseModel):
tool: str
ok: bool
ref: str
message: str = ""
data: Optional[Dict] = None
# adapters.py (demo logic)
from tools import *
from datetime import datetime
EQUIPMENT_POLICY = {
"Finance": ["MacBook", "Monitor", "Keyboard"],
"Sales": ["Laptop", "Monitor", "Headset"]
}
ACCESS_POLICY = ["email", "HR system", "Slack"]
def collect_personal_data(a: CollectPersonalDataArgs) -> ToolReceipt:
return ToolReceipt(tool="CollectPersonalData", ok=True, ref=f"data-{a.employee_id}",
message="Personal data collected", data={"employee_id": a.employee_id})
def verify_policy_acknowledgment(a: VerifyPolicyAcknowledgmentArgs) -> ToolReceipt:
required_policies = ["confidentiality agreement", "data protection policy"]
missing_policies = [policy for policy in required_policies if policy not in a.policies_acknowledged]
if missing_policies:
return ToolReceipt(tool="VerifyPolicyAcknowledgment", ok=False, ref="policies-missing",
message=f"Missing required policies: {', '.join(missing_policies)}")
return ToolReceipt(tool="VerifyPolicyAcknowledgment", ok=True, ref="policies-ok", message="Policies acknowledged")
def assign_equipment(a: AssignEquipmentArgs) -> ToolReceipt:
department_equipment = EQUIPMENT_POLICY.get(a.department)
if not department_equipment:
return ToolReceipt(tool="AssignEquipment", ok=False, ref="equip-policy-error", message="No equipment policy found for this department")
return ToolReceipt(tool="AssignEquipment", ok=True, ref=f"equip-{a.employee_id}", data={"equipment": department_equipment})
def provision_system_access(a: ProvisionSystemAccessArgs) -> ToolReceipt:
missing_access = [tool for tool in ACCESS_POLICY if tool not in a.access_list]
if missing_access:
return ToolReceipt(tool="ProvisionSystemAccess", ok=False, ref="access-missing", message=f"Missing access: {', '.join(missing_access)}")
return ToolReceipt(tool="ProvisionSystemAccess", ok=True, ref=f"access-{a.employee_id}", data={"access_granted": ACCESS_POLICY})
Agent Loop (proposal → verification → execution → receipts)
# agent_onboarding.py
import uuid, json
from typing import Any, Dict, List
from tools import *
from adapters import *
ALLOWED_TOOLS = {"CollectPersonalData", "VerifyPolicyAcknowledgment", "AssignEquipment", "ProvisionSystemAccess"}
def new_idem() -> str:
return f"idem-{uuid.uuid4()}"
def verify_proposal(p: Dict[str, Any]) -> str:
required = {"name","args","preconditions","idempotency_key"}
if not required.issubset(p): return "Missing proposal fields"
if p["name"] not in ALLOWED_TOOLS: return "Tool not allowed"
return ""
def execute(p: Dict[str, Any]) -> ToolReceipt:
n, a = p["name"], p["args"]
if n == "CollectPersonalData": return collect_personal_data(CollectPersonalDataArgs(**a))
if n == "VerifyPolicyAcknowledgment": return verify_policy_acknowledgment(VerifyPolicyAcknowledgmentArgs(**a))
if n == "AssignEquipment": return assign_equipment(AssignEquipmentArgs(**a))
if n == "ProvisionSystemAccess": return provision_system_access(ProvisionSystemAccessArgs(**a))
return ToolReceipt(tool=n, ok=False, ref="none", message="Unknown tool")
# --- Model shim returning a plan per contract (replace with your LLM call) ---
def call_model(contract_yaml: str, claims: List[Dict[str,Any]], employee_data: Dict[str,Any]) -> Dict[str,Any]:
return {
"summary": f"Employee {employee_data['name']} onboarding review.",
"decision": "approve",
"employee_data": {
"employee_id": employee_data["employee_id"],
"name": employee_data["name"],
"department": employee_data["department"],
"equipment_assigned": ["MacBook", "Monitor", "Keyboard"],
"system_access": ["email", "HR system", "Slack"]
},
"citations": ["policy:equipment:role_based","policy:access:tools_required"],
"next_steps": ["Collect personal data", "Verify policy acknowledgment", "Assign equipment", "Provision system access"],
"tool_proposals": [
{"name":"CollectPersonalData","args":{"employee_id":employee_data["employee_id"], "name":employee_data["name"], "department":employee_data["department"], "email":employee_data["email"]},
"preconditions":"Collect necessary employee details.","idempotency_key": new_idem()},
{"name":"VerifyPolicyAcknowledgment","args":{"employee_id":employee_data["employee_id"], "policies_acknowledged":employee_data["policies_acknowledged"]},
"preconditions":"Ensure policies are acknowledged.","idempotency_key": new_idem()},
{"name":"AssignEquipment","args":{"employee_id":employee_data["employee_id"], "department":employee_data["department"]},
"preconditions":"Assign role-based equipment.","idempotency_key": new_idem()},
{"name":"ProvisionSystemAccess","args":{"employee_id":employee_data["employee_id"], "access_list":employee_data["access_list"]},
"preconditions":"Provide system access as per policy.","idempotency_key": new_idem()}
]
}
def render_response(model_json: Dict[str,Any], receipts: List[ToolReceipt]) -> str:
idx = {r.tool:r for r in receipts}
lines = [model_json["summary"], ""]
lines.append(f"Decision: {model_json['decision']}")
for clause in model_json["employee_data"]:
lines.append(f"{clause}: {model_json['employee_data'][clause]}")
lines.append("\nNext steps:")
for s in model_json["next_steps"]:
lines.append(f"• {s}")
if idx.get("AssignEquipment") and idx["AssignEquipment"].ok:
lines.append(f"\nEquipment assigned: {', '.join(idx['AssignEquipment'].data['equipment'])}")
if idx.get("ProvisionSystemAccess") and idx["ProvisionSystemAccess"].ok:
lines.append(f"System access provisioned: {', '.join(idx['ProvisionSystemAccess'].data['access_granted'])}")
lines.append("\nCitations: " + ", ".join(model_json["citations"]))
return "\n".join(lines)
def handle(employee_data: Dict[str,Any]) -> str:
contract = open("contracts/onboarding_v1.yaml").read()
claims: List[Dict[str,Any]] = [] # load real claims from policy
plan = call_model(contract, claims, employee_data)
receipts: List[ToolReceipt] = []
for prop in plan["tool_proposals"]:
reason = verify_proposal(prop)
if reason:
receipts.append(ToolReceipt(tool=prop["name"], ok=False, ref="blocked", message=reason)); continue
rec = execute(prop)
receipts.append(rec)
if not rec.ok and prop["name"] in {"ProvisionSystemAccess"}:
break
return render_response(plan, receipts)
if __name__ == "__main__":
example_employee_data = {
"employee_id":"E001",
"name":"John Doe",
"department":"Engineering",
"email":"[email protected]",
"policies_acknowledged":["confidentiality agreement", "data protection policy"],
"access_list":["email", "HR system", "Slack"],
"equipment_assigned":["MacBook", "Monitor", "Keyboard"]
}
print(handle(example_employee_data))
The Prompt You’d Send to the Model (concise and testable)
System:
You are EmployeeOnboardingAgent. Follow the contract:
- Ask once if employee_id, name, department, policies_acknowledged, equipment_assigned, or access_list is missing.
- Cite 1–2 claim_ids per factual sentence using provided claims.
- Propose tools; never assert success without a receipt.
- Output JSON with keys: summary, decision, employee_data, citations[], next_steps[], tool_proposals[].
Claims (eligible only):
[ ... JSON array of onboarding policy claims like above ... ]
User:
Onboard this new employee:
{"employee_id":"E001","name":"John Doe","department":"Engineering","email":"[email protected]",
"policies_acknowledged":["confidentiality agreement", "data protection policy"],
"access_list":["email", "HR system", "Slack"],"equipment_assigned":["MacBook", "Monitor", "Keyboard"]}
How to adapt quickly
Replace the mock HR system, policy acknowledgment, equipment provisioning, and system access functions with real backend APIs. Implement idempotency for provisioning actions, ensure freshness of claims (e.g., for policy changes or role updates), and use validators to check for missing information, incorrect formats, or failed policy matches. Ship the contract, policy bundle, and decoder settings behind a feature flag to enable rapid iteration and testing with canary deployment and rollback support.