From agent-almanac
Implements pharmaceutical serialisation and track-and-trace systems compliant with EU FMD, US DSCSA, and global regulations. Covers unique ID generation, aggregation hierarchy, EPCIS exchange, verification endpoints. Use for product launches, EMVS/NMVS integration, or EPCIS repositories.
npx claudepluginhub pjt222/agent-almanacThis skill uses the workspace's default tool permissions.
---
Implements electronic signatures compliant with 21 CFR Part 11 Subpart C and EU Annex 11 for GxP records, covering manifestation, record binding, biometric/non-biometric controls, and policies. Use for regulated workflows like batch release and approvals.
Generates GDPR Article 30(1) RoPA for data controllers with all 7 mandatory fields including Python automation script. Useful for compliance, processing records, data mapping.
Prepares ISO 13485 certification documentation for medical device QMS via gap analysis, templates for Quality Manuals, procedures, work instructions, and Medical Device Files. Use for compliance audits, regulatory prep, or identifying gaps.
Share bugs, ideas, or general feedback.
Set up pharmaceutical serialisation systems for regulatory compliance with global track-and-trace mandates.
| Regulation | Region | Key Requirements | Deadline |
|---|---|---|---|
| EU FMD (2011/62/EU) | EU/EEA | Unique identifier + tamper-evident feature on each unit | Live since Feb 2019 |
| DSCSA | USA | Electronic, interoperable tracing at package level | Full enforcement Nov 2024+ |
| China NMPA | China | Unique drug traceability code per minimum saleable unit | Rolling |
| Brazil ANVISA (SNCM) | Brazil | Serialisation of pharmaceuticals with IUM | Rolling |
| Russia MDLP | Russia | Crypto-code per unit, mandatory scanning | Live |
Key data elements per regulation:
EU FMD unique identifier (per Delegated Regulation 2016/161):
DSCSA transaction information:
Expected: Clear understanding of which regulations apply to each product-market combination. On failure: Engage regulatory affairs to confirm market requirements before proceeding.
-- Core serialisation data model
CREATE TABLE serial_numbers (
id BIGSERIAL PRIMARY KEY,
gtin VARCHAR(14) NOT NULL, -- GS1 GTIN-14
serial_number VARCHAR(20) NOT NULL, -- Unique per GTIN
batch_lot VARCHAR(20) NOT NULL,
expiry_date DATE NOT NULL,
status VARCHAR(20) DEFAULT 'ACTIVE', -- ACTIVE, DECOMMISSIONED, DISPENSED, etc.
created_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(gtin, serial_number)
);
-- Aggregation hierarchy
CREATE TABLE aggregation (
id BIGSERIAL PRIMARY KEY,
parent_code VARCHAR(50) NOT NULL, -- SSCC or higher-level code
parent_level VARCHAR(10) NOT NULL, -- CASE, PALLET, BUNDLE
child_code VARCHAR(50) NOT NULL, -- GTIN+serial or child SSCC
child_level VARCHAR(10) NOT NULL, -- UNIT, BUNDLE, CASE
aggregation_event_id UUID NOT NULL,
created_at TIMESTAMPTZ DEFAULT NOW()
);
-- EPCIS events
CREATE TABLE epcis_events (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
event_type VARCHAR(30) NOT NULL, -- ObjectEvent, AggregationEvent, TransactionEvent
action VARCHAR(10) NOT NULL, -- ADD, OBSERVE, DELETE
biz_step VARCHAR(100), -- urn:epcglobal:cbv:bizstep:commissioning
disposition VARCHAR(100), -- urn:epcglobal:cbv:disp:active
read_point VARCHAR(100), -- urn:epc:id:sgln:location
event_time TIMESTAMPTZ NOT NULL,
event_timezone VARCHAR(6) NOT NULL,
payload JSONB NOT NULL,
created_at TIMESTAMPTZ DEFAULT NOW()
);
Aggregation hierarchy:
Pallet (SSCC)
└── Case (SSCC)
└── Bundle (GTIN + serial) [optional level]
└── Unit (GTIN + serial)
Expected: Data model supports full pack hierarchy with EPCIS event tracking. On failure: If existing ERP schema conflicts, design an integration layer rather than modifying ERP directly.
import secrets
import string
def generate_serial_number(length: int = 20, charset: str = None) -> str:
"""Generate a random serial number compliant with GS1 standards.
EU FMD requires randomised serial numbers to prevent prediction.
Max 20 characters, alphanumeric (GS1 Application Identifier 21).
"""
if charset is None:
# GS1 AI(21) allows: digits, uppercase, lowercase, and some special chars
# Most implementations use alphanumeric only for interoperability
charset = string.ascii_uppercase + string.digits
return ''.join(secrets.choice(charset) for _ in range(length))
def generate_serial_batch(gtin: str, batch_lot: str, expiry: str, count: int) -> list:
"""Generate a batch of unique serial numbers for a production run."""
serials = set()
while len(serials) < count:
serials.add(generate_serial_number())
return [
{
"gtin": gtin,
"serial_number": sn,
"batch_lot": batch_lot,
"expiry_date": expiry,
"status": "COMMISSIONED"
}
for sn in serials
]
Expected: Serial numbers are cryptographically random, unique per GTIN, and stored before printing. On failure: If a uniqueness collision occurs, regenerate the conflicting serial and log the event.
The 2D DataMatrix barcode encodes the GS1 element string:
(01)GTIN(21)Serial(10)Batch(17)Expiry
Example:
(01)05012345678901(21)A1B2C3D4E5(10)LOT123(17)261231
Where:
The GS1 DataMatrix uses FNC1 as a separator (GS character, ASCII 29) between variable-length fields.
def encode_gs1_element_string(gtin: str, serial: str, batch: str, expiry: str) -> str:
"""Encode GS1 element string for DataMatrix printing.
FNC1 (GS character \\x1d) separates variable-length fields.
AI(01) and AI(17) are fixed length, so no separator needed after them.
AI(21) and AI(10) are variable length and need FNC1 terminator.
"""
GS = '\x1d' # GS1 FNC1 / Group Separator
return f"01{gtin}21{serial}{GS}10{batch}{GS}17{expiry}"
Expected: Encoded strings verified by scanning test prints with a GS1-certified verifier (ISO 15415 grade C or above). On failure: If scan verification fails, check print quality, quiet zones, and encoding order.
MAH → Upload serial data → EU Hub → Distribute to National Systems (NMVS)
├── Germany (securPharm)
├── France (CTS)
├── Italy (AIFA)
└── ... 31 markets
API operations:
Trading Partner A → VRS Request → Verification Router → MAH's VRS → Response
Implement a VRS responder endpoint:
# Simplified VRS endpoint (DSCSA verification)
from fastapi import FastAPI, HTTPException
app = FastAPI()
@app.get("/verify/{gtin}/{serial}/{lot}/{expiry}")
async def verify_product(gtin: str, serial: str, lot: str, expiry: str):
"""DSCSA product verification endpoint."""
record = await lookup_serial(gtin, serial)
if record is None:
return {"verified": False, "reason": "SERIAL_NOT_FOUND"}
if record.batch_lot != lot or str(record.expiry_date) != expiry:
return {"verified": False, "reason": "DATA_MISMATCH"}
if record.status != "ACTIVE":
return {"verified": False, "reason": f"STATUS_{record.status}"}
return {"verified": True, "status": record.status}
Expected: Verification endpoints respond within 1 second with correct status. On failure: If national system upload fails, retry with exponential backoff and alert operations.
Record supply chain events in EPCIS 2.0 format:
{
"@context": "https://ref.gs1.org/standards/epcis/2.0.0/epcis-context.jsonld",
"type": "ObjectEvent",
"eventTime": "2025-03-15T10:30:00.000+01:00",
"eventTimeZoneOffset": "+01:00",
"epcList": ["urn:epc:id:sgtin:5012345.067890.A1B2C3D4E5"],
"action": "ADD",
"bizStep": "urn:epcglobal:cbv:bizstep:commissioning",
"disposition": "urn:epcglobal:cbv:disp:active",
"readPoint": {"id": "urn:epc:id:sgln:5012345.00001.0"},
"bizLocation": {"id": "urn:epc:id:sgln:5012345.00001.0"}
}
Key business steps in the pharma supply chain:
commissioning — serial number assigned to physical unitpacking — aggregation into cases/palletsshipping — departure from a locationreceiving — arrival at a locationdispensing — supplied to patient (decommission trigger)Expected: Every status change generates an EPCIS event with correct timestamps and locations. On failure: Failed event capture must be queued and retried; never silently dropped.
perform-csv-assessment — validate serialisation system as a computerised systemconduct-gxp-audit — audit serialisation processesimplement-audit-trail — audit trail for serialisation eventsserialize-data-formats — general data serialisation (different domain, complementary concepts)design-serialization-schema — schema design for data exchange formats