# Pet Farm Biosafety Monitor Agent Skill > A Claude skill from dungnotnull/pet-farm-biosafety-monitor-agent-skill. - **Type:** Skill - **Install:** `agentstack add skill-dungnotnull-pet-farm-biosafety-monitor-agent-skill-pet-farm-biosafety-monitor-agent-skill` - **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior - **Seller:** [dungnotnull](https://agentstack.voostack.com/s/dungnotnull) - **Installs:** 0 - **Category:** [Agent Skills](https://agentstack.voostack.com/c/agent-skills) - **Latest version:** 0.1.0 - **License:** MIT - **Upstream author:** [dungnotnull](https://github.com/dungnotnull) - **Source:** https://github.com/dungnotnull/pet-farm-biosafety-monitor-agent-skill ## Install ```sh agentstack add skill-dungnotnull-pet-farm-biosafety-monitor-agent-skill-pet-farm-biosafety-monitor-agent-skill ``` Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent. ## About # 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) ```python @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) ```python 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) ```python @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) ```json { "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 ```python 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 ```python 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 ```python 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) ```python 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 ```python 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 ```json { "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` | ```python 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). ```python 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. ```python 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 ```python 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: ```bash 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](https://github.com/dungnotnull) - **Source:** [dungnotnull/pet-farm-biosafety-monitor-agent-skill](https://github.com/dungnotnull/pet-farm-biosafety-monitor-agent-skill) - **License:** MIT Install and usage instructions live in the source repository linked above. ## Pricing - **Free** — Free ## Security capabilities Automated source analysis of v0.1.0 — what this tool can access: - **Network access:** no - **Filesystem access:** no - **Shell / process execution:** no - **Environment & secrets:** no - **Dynamic code execution:** no *"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.* ## Versions - **0.1.0** — security scan: passed — Imported from the upstream source. ## Links - Listing page: https://agentstack.voostack.com/l/skill-dungnotnull-pet-farm-biosafety-monitor-agent-skill-pet-farm-biosafety-monitor-agent-skill - Seller: https://agentstack.voostack.com/s/dungnotnull - Browse the marketplace: https://agentstack.voostack.com/browse --- Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.