Duty Of Care¶

Duty of Care — mvp.duty_of_care

Cluster: Uncategorised | Type: component | MCP Tools: 28

Overview¶

Duty-of-care guardrail framework for AI systems operating in wellbeing, vulnerable-user, or high-risk support contexts. Manages consent-gated sessions, risk assessment signals (distress, suicidal ideation, protective factors), structured wellbeing exercises (CBT/DBT/ACT-informed), and crisis lockout with emergency resource routing.

Pilot guardrail, not a clinical product

duty_of_care is suitable for MVP safety demonstrations, local MCP pilots, and supervised wellbeing-support workflows. It is not, by itself, a regulated clinical decision-support system, medical device, diagnostic tool, therapy provider, emergency service, or proof of clinical safety. Do not make clinical, therapeutic, diagnostic, or regulated healthcare claims for this component without a formal intended-use statement, jurisdiction-specific regulatory classification, clinical governance, privacy/security review, quality-management controls, validation evidence, and human clinical oversight.

When to use:

Building AI companions or support agents that interact with vulnerable users
Enforcing duty-of-care guardrails (crisis lock, escalation thresholds) in agent pipelines
Providing structured wellbeing micro-interventions (thought records, DBT skills) with appropriate disclaimers and review

Example:

from mvp.duty_of_care import DutyCareBlock, DutyCareInput

block = DutyCareBlock(name="duty_of_care")
result = block.infer(DutyCareInput(
    op="session_begin", user_id="u123", consent_given=True
))
# result.ok → True; result.value → DutyCareOutput with session_id, tier, disclaimer

Works well with: align_health, professional_standards, immune_system

Public API¶

`DutyCareBlock(AIBlock[DutyCareInput, DutyCareOutput, None])`¶

Tier 1 duty-of-care block. Exposes first_aid-tier ops only.

Field	Type	Default
`name`	`str`	'duty_of_care'
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`db_path`	`str`	field(default_factory=lambda: os.environ.get('DUTY_CARE_DB_PATH', str(Path.home() / '.g6' / 'duty_of_care.db')))

Methods:

`infer(data: DutyCareInput) -> Result[DutyCareOutput]`¶

`DutyCarePatternRuntime`¶

Deterministic crisis-envelope self-check over an emitted output (no LLM, no I/O).

Methods:

`validate_output(result: Any) -> list[str]`¶

Return crisis-envelope / internal-consistency violations of an emitted output.

`assert_consistent(result: Any) -> Any`¶

Live-path guard: raise if the emitted output is a crisis-hiding envelope.

`DutyCareInput(BaseModel)`¶

Tier 1 input for DutyCareBlock (simplified interface).

Field	Type	Default
`op`	`Literal['session_begin', 'session_end', 'session_status', 'risk_assess', 'dbt_distress_tolerance', 'crisis_lock', 'info']`	required
`user_id`	`str`	''
`session_id`	`str`	''
`consent_given`	`bool`	False
`closing_notes`	`str`	''
`distress_level`	`int`	0
`suicidal_ideation`	`bool`	False
`self_harm_current`	`bool`	False
`protective_factors`	`list[str]`	Field(default_factory=list)
`skill_type`	`str`	''
`crisis_reason`	`str`	''
`locale`	`str`	'AU'

`DutyCareOutput(BaseModel)`¶

Tier 1 output from DutyCareBlock.

Field	Type	Default
`op`	`str`	required
`success`	`bool`	True
`session_id`	`str`	''
`tier`	`str`	''
`risk_level`	`str`	''
`crisis_locked`	`bool`	False
`disclaimer`	`str`	''
`message`	`str`	''
`warning`	`str`	''
`content`	`dict[str, Any]`	Field(default_factory=dict)
`resources`	`list[dict[str, str]]`	Field(default_factory=list)
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None

`DutyCareSkill`¶

Field	Type	Default
`name`	`str`	required
`pattern_slug`	`str`	required
`description`	`str`	required
`executable`	`bool`	required
`mechanism`	`str`	required
`capabilities`	`tuple[str, ...]`	required
`triggers`	`tuple[str, ...]`	required
`risk_notes`	`tuple[str, ...]`	required

Methods:

`compact() -> dict[str, Any]`¶

`DutyCareSkillCatalog`¶

Maps each applied pattern slug to a duty_of_care deterministic crisis-safety skill.

Methods:

`list_skills() -> list[DutyCareSkill]`¶

`executable_skills() -> list[DutyCareSkill]`¶

`get(slug: str) -> DutyCareSkill | None`¶

`DutyCareMCPBlock(AIBlock[MCPDutyCareInput, MCPDutyCareOutput, dict])`¶

28-op duty-of-care MCP block with tier gating + crisis check.

Field	Type	Default
`name`	`str`	'duty_care_mcp'
`state`	`dict`	field(default_factory=dict)
`db_path`	`str`	':memory:'
`resource_bounds`	`ResourceBounds`	field(default_factory=ResourceBounds)
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`infer(data: MCPDutyCareInput) -> Result[MCPDutyCareOutput]`¶

`list_patterns() -> dict[str, Any]`¶

Surface the applied deterministic crisis-safety pattern catalog (read-only).

`Tier(str, Enum)`¶

Duty-of-care access tiers (ascending privilege).

`RiskLevel(str, Enum)`¶

Traffic-light risk state machine.

`MCPDutyCareRecord(BaseModel)`¶

Field	Type	Default
`id`	`str`	required
`record_type`	`str`	required
`key`	`str`	required
`value`	`str`	required
`tags`	`list[str]`	Field(default_factory=list)
`timestamp`	`str`	''
`metadata`	`dict[str, Any]`	Field(default_factory=dict)

`MCPDutyCareInput(BaseModel)`¶

Field	Type	Default
`op`	`Literal['session_begin', 'session_end', 'session_status', 'tier_upgrade', 'session_history', 'risk_assess', 'risk_history', 'risk_override', 'screen_administer', 'screen_score', 'screen_trend', 'screen_list', 'cbt_thought_record', 'cbt_restructure', 'cbt_socratic', 'cbt_log', 'dbt_distress_tolerance', 'dbt_emotion_regulation', 'dbt_log', 'act_values', 'act_acceptance', 'act_log', 'crisis_lock', 'crisis_unlock', 'crisis_resources_get', 'crisis_resources_update', 'info', 'capabilities']`	required
`session_id`	`str`	''
`user_id`	`str`	''
`consent_given`	`bool`	False
`closing_notes`	`str`	''
`target_tier`	`str`	''
`clinician_approved`	`bool`	False
`distress_level`	`int`	0
`suicidal_ideation`	`bool`	False
`self_harm_current`	`bool`	False
`protective_factors`	`list[str]`	Field(default_factory=list)
`risk_notes`	`str`	''
`new_risk_level`	`str`	''
`clinician_id`	`str`	''
`override_reason`	`str`	''
`instrument`	`str`	''
`responses`	`list[int]`	Field(default_factory=list)
`situation`	`str`	''
`automatic_thought`	`str`	''
`emotions`	`list[str]`	Field(default_factory=list)
`emotion_intensity`	`int`	0
`evidence_for`	`list[str]`	Field(default_factory=list)
`evidence_against`	`list[str]`	Field(default_factory=list)
`balanced_thought`	`str`	''
`thought_id`	`str`	''
`skill_type`	`str`	''
`skill_details`	`dict[str, Any]`	Field(default_factory=dict)
`effectiveness`	`int`	0
`values_domain`	`str`	''
`current_alignment`	`int`	0
`desired_alignment`	`int`	0
`exercise_type`	`str`	''
`exercise_response`	`str`	''
`crisis_reason`	`str`	''
`locale`	`str`	'AU'
`resources_payload`	`list[dict[str, str]]`	Field(default_factory=list)
`limit`	`int`	50
`query`	`str`	''

`MCPDutyCareOutput(BaseModel)`¶

Field	Type	Default
`op`	`str`	required
`success`	`bool`	True
`session_id`	`str`	''
`user_id`	`str`	''
`tier`	`str`	''
`risk_level`	`str`	''
`crisis_locked`	`bool`	False
`consent_required`	`bool`	False
`disclaimer`	`str`	''
`instrument`	`str`	''
`items`	`list[str]`	Field(default_factory=list)
`total_score`	`int`	0
`severity`	`str`	''
`interpretation`	`str`	''
`subscales`	`dict[str, Any]`	Field(default_factory=dict)
`content`	`dict[str, Any]`	Field(default_factory=dict)
`resources`	`list[dict[str, str]]`	Field(default_factory=list)
`records`	`list[dict[str, Any]]`	Field(default_factory=list)
`count`	`int`	0
`message`	`str`	''
`warning`	`str`	''
`metadata`	`dict[str, Any]`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None

`DutyCareStore`¶

Sync SQLite duty-of-care store with 7 tables.

Constructor:

Parameter	Type	Default
`db_path`	`str`	':memory:'

Methods:

`create_session(user_id: str, consent_given: bool, disclaimer_text: str) -> dict`¶

`get_session(session_id: str) -> dict | None`¶

`end_session(session_id: str, closing_notes: str = '') -> bool`¶

`update_session_tier(session_id: str, tier: str) -> None`¶

`update_session_risk(session_id: str, risk_level: str) -> None`¶

`update_session_consent(session_id: str) -> None`¶

`set_crisis_locked(session_id: str, locked: bool) -> None`¶

`get_user_sessions(user_id: str, limit: int = 50) -> list[dict]`¶

`get_user_highest_consented_tier(user_id: str) -> str | None`¶

Return the highest tier from consented sessions for a user.

`count_user_sessions(user_id: str) -> int`¶

`record_risk(session_id: str, user_id: str, previous_level: str, new_level: str, distress_level: int = 0, suicidal_ideation: bool = False, self_harm_current: bool = False, protective_factors: list[str] | None = None, risk_notes: str = '', auto_triggered: bool = False) -> str`¶

`get_risk_history(user_id: str, limit: int = 50) -> list[dict]`¶

`record_screening(session_id: str, user_id: str, instrument: str, responses: list[int], total_score: int, severity: str, subscales: dict, interpretation: str) -> str`¶

`get_screening_trend(user_id: str, instrument: str, limit: int = 20) -> list[dict]`¶

`list_screenings(user_id: str, session_id: str = '', limit: int = 50) -> list[dict]`¶

`count_user_screenings(user_id: str) -> int`¶

`record_intervention(session_id: str, user_id: str, modality: str, intervention_type: str, content: dict, effectiveness: int = 0) -> str`¶

`get_interventions(user_id: str, modality: str = '', session_id: str = '', limit: int = 50) -> list[dict]`¶

`record_crisis_event(session_id: str, user_id: str, event_type: str, reason: str = '', clinician_id: str = '', resources: list[dict] | None = None, risk_assess_id: str = '') -> str`¶

`audit(session_id: str, user_id: str, op: str, tier: str = '', risk_level: str = '', detail: dict | None = None) -> None`¶

`count_all() -> dict[str, int]`¶

Functions¶

`applied_agentic_patterns() -> list[dict[str, Any]]`¶

Return compact metadata for duty_of_care-applied vendored patterns.

`get_skill_catalog() -> DutyCareSkillCatalog`¶

MCP Tools¶

Operation	Source
`session_begin`	duty_care_mcp
`session_end`	duty_care_mcp
`session_status`	duty_care_mcp
`tier_upgrade`	duty_care_mcp
`session_history`	duty_care_mcp
`risk_assess`	duty_care_mcp
`risk_history`	duty_care_mcp
`risk_override`	duty_care_mcp
`screen_administer`	duty_care_mcp
`screen_score`	duty_care_mcp
`screen_trend`	duty_care_mcp
`screen_list`	duty_care_mcp
`cbt_thought_record`	duty_care_mcp
`cbt_restructure`	duty_care_mcp
`cbt_socratic`	duty_care_mcp
`cbt_log`	duty_care_mcp
`dbt_distress_tolerance`	duty_care_mcp
`dbt_emotion_regulation`	duty_care_mcp
`dbt_log`	duty_care_mcp
`act_values`	duty_care_mcp
`act_acceptance`	duty_care_mcp
`act_log`	duty_care_mcp
`crisis_lock`	duty_care_mcp
`crisis_unlock`	duty_care_mcp
`crisis_resources_get`	duty_care_mcp
`crisis_resources_update`	duty_care_mcp
`info`	duty_care_mcp
`capabilities`	duty_care_mcp

Duty Of Care¶

Overview¶

Public API¶

DutyCareBlock(AIBlock[DutyCareInput, DutyCareOutput, None])¶

infer(data: DutyCareInput) -> Result[DutyCareOutput]¶

DutyCarePatternRuntime¶

validate_output(result: Any) -> list[str]¶

assert_consistent(result: Any) -> Any¶

DutyCareInput(BaseModel)¶

DutyCareOutput(BaseModel)¶

DutyCareSkill¶

compact() -> dict[str, Any]¶

DutyCareSkillCatalog¶

list_skills() -> list[DutyCareSkill]¶

executable_skills() -> list[DutyCareSkill]¶

get(slug: str) -> DutyCareSkill | None¶

DutyCareMCPBlock(AIBlock[MCPDutyCareInput, MCPDutyCareOutput, dict])¶

infer(data: MCPDutyCareInput) -> Result[MCPDutyCareOutput]¶

list_patterns() -> dict[str, Any]¶

Tier(str, Enum)¶

RiskLevel(str, Enum)¶

MCPDutyCareRecord(BaseModel)¶

MCPDutyCareInput(BaseModel)¶

MCPDutyCareOutput(BaseModel)¶

DutyCareStore¶

create_session(user_id: str, consent_given: bool, disclaimer_text: str) -> dict¶

get_session(session_id: str) -> dict | None¶

end_session(session_id: str, closing_notes: str = '') -> bool¶

update_session_tier(session_id: str, tier: str) -> None¶

update_session_risk(session_id: str, risk_level: str) -> None¶

update_session_consent(session_id: str) -> None¶

set_crisis_locked(session_id: str, locked: bool) -> None¶

get_user_sessions(user_id: str, limit: int = 50) -> list[dict]¶

get_user_highest_consented_tier(user_id: str) -> str | None¶

count_user_sessions(user_id: str) -> int¶

record_risk(session_id: str, user_id: str, previous_level: str, new_level: str, distress_level: int = 0, suicidal_ideation: bool = False, self_harm_current: bool = False, protective_factors: list[str] | None = None, risk_notes: str = '', auto_triggered: bool = False) -> str¶

get_risk_history(user_id: str, limit: int = 50) -> list[dict]¶

record_screening(session_id: str, user_id: str, instrument: str, responses: list[int], total_score: int, severity: str, subscales: dict, interpretation: str) -> str¶

get_screening_trend(user_id: str, instrument: str, limit: int = 20) -> list[dict]¶

list_screenings(user_id: str, session_id: str = '', limit: int = 50) -> list[dict]¶

count_user_screenings(user_id: str) -> int¶

record_intervention(session_id: str, user_id: str, modality: str, intervention_type: str, content: dict, effectiveness: int = 0) -> str¶

get_interventions(user_id: str, modality: str = '', session_id: str = '', limit: int = 50) -> list[dict]¶

record_crisis_event(session_id: str, user_id: str, event_type: str, reason: str = '', clinician_id: str = '', resources: list[dict] | None = None, risk_assess_id: str = '') -> str¶

audit(session_id: str, user_id: str, op: str, tier: str = '', risk_level: str = '', detail: dict | None = None) -> None¶

count_all() -> dict[str, int]¶

Functions¶

applied_agentic_patterns() -> list[dict[str, Any]]¶

get_skill_catalog() -> DutyCareSkillCatalog¶

MCP Tools¶

`DutyCareBlock(AIBlock[DutyCareInput, DutyCareOutput, None])`¶

`infer(data: DutyCareInput) -> Result[DutyCareOutput]`¶

`DutyCarePatternRuntime`¶

`validate_output(result: Any) -> list[str]`¶

`assert_consistent(result: Any) -> Any`¶

`DutyCareInput(BaseModel)`¶

`DutyCareOutput(BaseModel)`¶

`DutyCareSkill`¶

`compact() -> dict[str, Any]`¶

`DutyCareSkillCatalog`¶

`list_skills() -> list[DutyCareSkill]`¶

`executable_skills() -> list[DutyCareSkill]`¶

`get(slug: str) -> DutyCareSkill | None`¶

`DutyCareMCPBlock(AIBlock[MCPDutyCareInput, MCPDutyCareOutput, dict])`¶

`infer(data: MCPDutyCareInput) -> Result[MCPDutyCareOutput]`¶

`list_patterns() -> dict[str, Any]`¶

`Tier(str, Enum)`¶

`RiskLevel(str, Enum)`¶

`MCPDutyCareRecord(BaseModel)`¶

`MCPDutyCareInput(BaseModel)`¶

`MCPDutyCareOutput(BaseModel)`¶

`DutyCareStore`¶

`create_session(user_id: str, consent_given: bool, disclaimer_text: str) -> dict`¶

`get_session(session_id: str) -> dict | None`¶

`end_session(session_id: str, closing_notes: str = '') -> bool`¶

`update_session_tier(session_id: str, tier: str) -> None`¶

`update_session_risk(session_id: str, risk_level: str) -> None`¶

`update_session_consent(session_id: str) -> None`¶

`set_crisis_locked(session_id: str, locked: bool) -> None`¶

`get_user_sessions(user_id: str, limit: int = 50) -> list[dict]`¶

`get_user_highest_consented_tier(user_id: str) -> str | None`¶

`count_user_sessions(user_id: str) -> int`¶

`record_risk(session_id: str, user_id: str, previous_level: str, new_level: str, distress_level: int = 0, suicidal_ideation: bool = False, self_harm_current: bool = False, protective_factors: list[str] | None = None, risk_notes: str = '', auto_triggered: bool = False) -> str`¶

`get_risk_history(user_id: str, limit: int = 50) -> list[dict]`¶

`record_screening(session_id: str, user_id: str, instrument: str, responses: list[int], total_score: int, severity: str, subscales: dict, interpretation: str) -> str`¶

`get_screening_trend(user_id: str, instrument: str, limit: int = 20) -> list[dict]`¶

`list_screenings(user_id: str, session_id: str = '', limit: int = 50) -> list[dict]`¶

`count_user_screenings(user_id: str) -> int`¶

`record_intervention(session_id: str, user_id: str, modality: str, intervention_type: str, content: dict, effectiveness: int = 0) -> str`¶

`get_interventions(user_id: str, modality: str = '', session_id: str = '', limit: int = 50) -> list[dict]`¶

`record_crisis_event(session_id: str, user_id: str, event_type: str, reason: str = '', clinician_id: str = '', resources: list[dict] | None = None, risk_assess_id: str = '') -> str`¶

`audit(session_id: str, user_id: str, op: str, tier: str = '', risk_level: str = '', detail: dict | None = None) -> None`¶

`count_all() -> dict[str, int]`¶

`applied_agentic_patterns() -> list[dict[str, Any]]`¶

`get_skill_catalog() -> DutyCareSkillCatalog`¶