Pet Farm Biosafety Monitor Agent Skill

SKILL.md - Skill Registry & Agent Architecture

pet-farm-biosafety-monitor v3.0.0

This document is the canonical reference for the skill registry that powers the modular agent layer of pet-farm-biosafety-monitor. It explains how skills are registered, resolved, executed and validated, the JSON schemas each skill and tool declares, the lifecycle hooks/event bus, and how to extend the registry with new skills or tools.

> The registry lives in pet_farm_biosafety/agents/ (skills) and > pet_farm_biosafety/tools/ (tools). Machine-readable specs are exported to > assets/schemas/skills.json and assets/schemas/tools.json via > python scripts/export_schemas.py.

---

1. Architecture at a glance

AgentOrchestrator
   |
   +-- SkillRegistry  (register / resolve / execute / validate)
   |      +-- requirements          (Step 1)
   |      +-- evidence_collector    (Step 2)
   |      +-- core_analysis         (Step 3)
   |      +-- knowledge_query       (Step 4)
   |      +-- advisor               (Step 5)
   |      +-- quality_gate          (Step 6)
   |
   +-- ToolRegistry   (schema-defined agent tools)
   |      +-- web_search, web_fetch
   |      +-- knowledge_query, knowledge_update
   |      +-- pathogen_risk, disinfection, quarantine, compliance
   |      +-- token_budget
   |
   +-- HookManager / EventBus  (lifecycle + observability)
   |      +-- LoggingHook, MetricsHook, TokenBudgetHook, DegradationHook
   |
   +-- TokenBudgeter / ContextWindowManager (context-window discipline)
   |
   +-- ChainRouter (ordered execution + graceful fallback)

A shared AgentContext carries state, the tool registry, the hook manager, the event bus and the token budgeter between skills, so skills stay stateless and cross-skill synchronisation is explicit.

---

2. Core contracts

2.1 SkillSpec (declarative metadata + schemas)

@dataclass
class SkillSpec:
    name: str
    description: str
    input_schema: Dict[str, Any]   # JSON-schema-style
    output_schema: Dict[str, Any]  # JSON-schema-style
    version: str = "1.0.0"
    tags: List[str] = []
    timeout_seconds: Optional[float] = None
    fallback_skill: Optional[str] = None

2.2 SkillBase (the contract every skill honours)

class SkillBase:
    spec: Optional[SkillSpec] = None
    def run(self, ctx: AgentContext) -> Any: ...      # main work
    def fallback(self, ctx, exc) -> Any: ...           # graceful fallback
    def serialise(self, payload) -> Any: ...           # payload -> schema shape
    def execute(self, ctx: AgentContext) -> AgentResult  # timing + error handling

execute() wraps run() with timing, status mapping (SUCCESS / FALLBACK / SKIPPED / FAILED) and fallback handling, and emits skill.start / skill.end events.

2.3 AgentContext (shared state)

@dataclass
class AgentContext:
    user_input: str
    options: Dict[str, Any]
    state: Dict[str, Any]            # each skill's AgentResult, keyed by name
    tool_registry: Optional[ToolRegistry]
    hook_manager: Optional[HookManager]
    event_bus: Optional[EventBus]
    token_budgeter: Optional[TokenBudgeter]
    metadata: Dict[str, Any]

2.4 AgentResult (serialisable return value)

{
  "skill_name": "core_analysis",
  "status": "success",
  "payload": { "...rich domain object via serialise()..." },
  "duration_seconds": 0.0021,
  "error": null,
  "fallback_used": null,
  "run_id": "a1b2c3...",
  "started_at": "2026-07-27T10:00:00Z"
}

---

3. Registration

3.1 Register the default chain

from pet_farm_biosafety.agents import get_registry, register_default_skills
registry = get_registry()
register_default_skills(registry)   # registers all 6 skills (override=True)

3.2 Register a custom skill

from pet_farm_biosafety.agents import SkillBase, SkillSpec, AgentContext

class MySkill(SkillBase):
    spec = SkillSpec(
        name="my_skill",
        description="Does a thing.",
        input_schema={"type": "object", "required": ["x"], "properties": {"x": {"type": "string"}}},
        output_schema={"type": "object", "required": ["y"], "properties": {"y": {"type": "string"}}},
        tags=["custom"],
        fallback_skill=None,
    )
    def run(self, ctx: AgentContext) -> dict:
        return {"y": ctx.options.get("x", "")}
    def serialise(self, payload):
        return payload

registry.register(MySkill(), override=True)

3.3 Resolution

• registry.get(name) -> SkillBase (raises SkillError if missing).
• registry.resolve_by_tag(tag) -> all skills with that tag.
• registry.has(name), registry.names(), registry.specs().

---

4. Execution & validation

4.1 Execute a single skill

ctx = AgentContext(user_input="...", options={...}, tool_registry=tools)
result = registry.execute("core_analysis", ctx, validate=True)

When validate=True, the registry calls skill.serialise(result.payload) and runs a lightweight JSON-schema validator (validate_against_schema) against spec.output_schema; schema errors are surfaced in result.error without discarding the payload.

4.2 Execute a chain (ChainRouter)

from pet_farm_biosafety.agents import ChainRouter
router = ChainRouter(["requirements", "evidence_collector", "core_analysis",
                      "knowledge_query", "advisor", "quality_gate"], registry)
report = router.run(ctx)

The router stores each AgentResult on ctx.state[skill_name], so downstream skills read upstream outputs via ctx.get("requirements").payload. A FAILED skill aborts the chain unless it declared a fallback_skill, in which case the router executes that fallback once.

4.3 Validation only

errors = validate_against_schema(payload, spec.output_schema)  # [] == valid

---

5. Skill catalog (default chain)

| # | Skill name | Tags | Input (required) | Output (required) | |---|------------|------|------------------|-------------------| | 1 | requirements | intake, step1 | user_input | object_of_analysis, language | | 2 | evidence_collector | data, step2 | requirements | degradation_level, total_sources | | 3 | core_analysis | domain, step3 | requirements, evidence | overall_risk, pathogen_risks | | 4 | knowledge_query | knowledge, step4 | keywords | coverage_rating, citations | | 5 | advisor | synthesis, step5 | core_analysis, evidence, knowledge | verdict | | 6 | quality_gate | gate, step6, final | all upstream | all_gates_passed, gates_passed, gates_total, verdict |

Full schemas (input + output JSON) are in assets/schemas/skills.json.

Example: core_analysis output schema

{
  "type": "object",
  "required": ["overall_risk", "pathogen_risks"],
  "properties": {
    "overall_risk": {"type": "string"},
    "pathogen_risks": {"type": "array"},
    "quarantine_period_days": {"type": "integer"},
    "disinfectants": {"type": "array"},
    "compliance": {"type": "object"},
    "scenarios": {"type": "array"},
    "degradation_level": {"type": "string"},
    "limitations": {"type": "array", "items": {"type": "string"}}
  }
}

---

6. Tool registry

Tools are the bounded, schema-defined operations a skill may invoke. Each tool declares input_schema / output_schema (function-calling style) and returns a ToolResult with ok, data, error, duration_seconds.

| Tool | Input (required) | Output (key fields) | |------|------------------|---------------------| | web_search | query | results, count, degraded | | web_fetch | url | content, chars, degraded | | knowledge_query | keywords | citations, coverage_rating, gaps | | knowledge_update | - | added, dry_run, news_only | | pathogen_risk | species | risks, count | | disinfection | species | disinfectants, core_vaccines, gaps | | quarantine | species | quarantine_period_days, gaps, recommendations | | compliance | species | woah_compliant, aphis_compliant, ... | | token_budget | - | limit, consumed, remaining, utilisation_pct |

from pet_farm_biosafety.tools import get_tool_registry, register_domain_tools
tr = get_tool_registry()
register_domain_tools(tr)
result = tr.invoke("pathogen_risk", {"species": ["dogs"]})
assert result.ok and result.data["count"] >= 1

Full tool specs: assets/schemas/tools.json.

---

7. Hooks & event bus

Lifecycle events emitted by the framework:

| Event | Payload | Emitted by | |-------|---------|------------| | chain.start | chain (list) | ChainRouter | | chain.step | skill, index | ChainRouter | | chain.fallback | from_skill, to_skill | ChainRouter | | chain.end | completed, failed, skipped, aborted_at | ChainRouter | | skill.start | skill | SkillBase.execute | | skill.end | skill, status, duration | SkillBase.execute |

Built-in hooks: LoggingHook (structured event log), MetricsHook (event counts + skill/chain durations), TokenBudgetHook (soft/hard budget), DegradationHook (worst degradation level).

from pet_farm_biosafety.hooks import HookManager, EventBus, MetricsHook
bus = EventBus()
mgr = HookManager(bus)
metrics = MetricsHook(); mgr.register(metrics)
bus.emit("skill.end", skill="core_analysis", status="success", duration=0.01)
print(metrics.snapshot().to_dict())

---

8. Context-window management

TokenBudgeter estimates tokens (dependency-free heuristic) and tracks consumption across the chain; ContextWindowManager adds priority-ordered items and evicts the lowest-priority ones when the budget is approached.

from pet_farm_biosafety.tools import TokenBudgeter, ContextWindowManager
b = TokenBudgeter(limit=8000)
cwm = ContextWindowManager(b)
cwm.add("user_input", long_text, priority=1)
cwm.add("evidence", evidence_text, priority=5)
print(b.report().to_dict())

The AgentOrchestrator wires a budgeter automatically when feature_flags.enable_token_budget is true and token_limit > 0.

---

9. End-to-end usage

from pet_farm_biosafety import AgentOrchestrator

orch = AgentOrchestrator()
result = orch.run("Analyze my dog breeding facility with 20 dogs", offline=True)
print(result.advisory.verdict.value)
print(result.all_gates_passed)

# With full observability:
result, report = orch.run("...", offline=True, return_report=True)
print(report.to_dict())

CLI equivalents:

python -m pet_farm_biosafety orchestrate "Analyze my dog breeding facility" --report
python -m pet_farm_biosafety skills --verbose
python -m pet_farm_biosafety tools --verbose
python -m pet_farm_biosafety config --file config/default.toml

---

10. Extending the registry

1. Subclass SkillBase, set a SkillSpec, implement run() and serialise().
2. Register via registry.register(MySkill(), override=True).
3. Add the skill name to your ChainRouter chain (or replace one).
4. Export schemas with python scripts/export_schemas.py.
5. Add unit tests under tests/unit/ mirroring the existing skill tests.

For tools: subclass Tool, set name/input_schema/output_schema, implement _run(), register via tool_registry.register(MyTool(), override=True).

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: dungnotnull
• Source: dungnotnull/pet-farm-biosafety-monitor-agent-skill
• License: MIT

Install and usage instructions live in the source repository linked above.