Skip to content

Job Allied Health¶

job_allied_health — G6 AlliedHealth job agent.

Cluster: Job Agents | Type: component | MCP Tools: 25

Overview¶

Domain-specialist job agent for allied health professionals including physiotherapists, occupational therapists, dietitians, and radiographers. Supports patient assessment, therapy plan development, progress tracking, clinical documentation, specialist referrals, and outcome analysis — all within a safety-first JobAgentBlock framework with mandatory human-review flags for high-stakes clinical decisions.

Clinical decision-support caveat

job_allied_health is an advisory workflow assistant, not an autonomous clinical system. The public JobAlliedHealthBlock and MCP server path apply strict safety gates to high-stakes operations such as treatment planning, specialist referral, outcome analysis, and intervention evaluation. Those operations require structured professional attestation before execution.

Low-level Python handler functions in allied_health_mcp.handlers are implementation internals and can be imported directly by code in the same process. They are not the safety boundary. Production use should route user-facing calls through JobAlliedHealthBlock or JobAlliedHealthMCPBlock, keep human review in the workflow, and avoid treating generated output as clinical advice or regulated compliance evidence.

When to use:

Generating personalised therapy plans from initial patient assessment data
Tracking rehabilitation progress across sessions and flagging plateau or regression
Drafting structured clinical notes compliant with healthcare documentation standards
Coordinating multi-disciplinary referrals when a patient's needs span specialties

Example:

from mvp.job_allied_health import JobAlliedHealthBlock, JobAlliedHealthInput

block = JobAlliedHealthBlock()
result = block.infer(JobAlliedHealthInput(
    task="Assess post-surgical shoulder mobility and draft a 6-week physiotherapy plan",
    context={"patient_id": "P-00412", "surgery": "rotator_cuff_repair", "weeks_post_op": 3},
))
# result.ok → True; result.value → JobAlliedHealthOutput with result, artifacts

Works well with: job_framework, job_medical_surgical, job_psychologist

Executable surface and backend truth:

The public executable MCP surface is the 25 ops in ALL_OPS / skill/SKILL.md.
The many files under skill/skills/ are advisory workflow references, not extra executable tools.
Patient records are currently backed by AlliedHealthStore through StoreBackedAdapter; no external EHR/EMR backend is configured unless a deployment supplies one.
Outputs use the canonical completion states verified, qualified-draft, and blocked-escalated; missing external EHR support is surfaced as qualified-draft capability evidence, not hidden.

Public API¶

`JobAlliedHealthBlock(JobAgentBlock)`¶

G6 Allied Health job agent with advisory agentic op routing.

Field	Type	Default
`name`	`str`	'job_allied_health'
`sector`	`SectorClassification`	field(default_factory=lambda: _SECTOR)
`toolkit`	`ToolkitSpec \\| None`	field(default_factory=lambda: JOB_TOOLKITS.get('allied_health'))
`mcp_module`	`str`	'mvp.job_allied_health.allied_health_mcp.server'
`agentic_planner`	`object \\| None`	None
`capabilities`	`ClassVar[set[type]]`	{Extensible, HumanLearnable, Collaborative, ProblemSolvable, KnowledgeGrounded, Memorable, AgentCommunicable}

Methods:

`infer(data)`¶

`JobAlliedHealthInput(JobInput)`¶

Input for the AlliedHealth job agent.

`JobAlliedHealthOutput(JobOutput)`¶

`JobAlliedHealthMCPBlock(AIBlock[MCPJobAlliedHealthInput, MCPJobAlliedHealthOutput, dict])`¶

25-op MCP block for the AlliedHealth job agent.

Field	Type	Default
`name`	`str`	'job_allied_health_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: MCPJobAlliedHealthInput) -> Result[MCPJobAlliedHealthOutput]`¶

`MCPJobAlliedHealthInput(BaseModel)`¶

Input to JobAlliedHealthMCPBlock - 25-op dispatch.

Field	Type	Default
`op`	`Literal['assess_patient', 'plan_therapy', 'track_progress', 'document_treatment', 'refer_specialist', 'analyze_outcomes', 'evaluate_intervention', 'assess_mobility', 'benchmark_recovery', 'audit_documentation', 'create_proposal', 'review_deliverable', 'delegate_task', 'report_status', 'request_feedback', 'store_artifact', 'retrieve_artifact', 'list_artifacts', 'search_artifacts', 'archive', 'plan_sprint', 'reflect_on_outcome', 'list_patterns', 'get_capabilities', 'info']`	required
`task`	`str`	''
`context`	`dict[str, Any]`	Field(default_factory=dict)
`parameters`	`dict[str, Any]`	Field(default_factory=dict)
`artifact_id`	`str`	''
`query`	`str`	''

`MCPJobAlliedHealthOutput(BaseModel)`¶

Output from JobAlliedHealthMCPBlock.

Field	Type	Default
`op`	`str`	required
`result`	`str`	''
`artifacts`	`list[dict[str, Any]]`	Field(default_factory=list)
`records`	`list[dict[str, Any]]`	Field(default_factory=list)
`message`	`str`	''
`count`	`int`	0
`found`	`bool`	False
`metadata`	`dict[str, Any]`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)

`AlliedHealthStore(JobStore)`¶

SQLite store for the AlliedHealth job agent.

Constructor:

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

Methods:

`add_patient(demographics: dict | None = None, referral_source: str = '', primary_condition: str = '', functional_status: str = '') -> str`¶

`get_patient(patient_id: str) -> dict | None`¶

`list_patients(limit: int = 100) -> list[dict]`¶

`add_assessment(patient_id: str, assessment_type: str = '', discipline: str = '', findings: dict | None = None, scores: dict | None = None, clinician: str = '') -> str`¶

`get_assessments(patient_id: str) -> list[dict]`¶

`add_treatment_plan(patient_id: str, diagnosis: str = '', goals: list | None = None, interventions: list | None = None, frequency: str = '', duration_weeks: int = 0) -> str`¶

`get_treatment_plans(patient_id: str) -> list[dict]`¶

`update_plan_status(plan_id: str, status: str) -> None`¶

`add_session(patient_id: str, plan_id: str = '', discipline: str = '', interventions: list | None = None, outcomes: dict | None = None, duration_mins: int = 0, session_date: str = '') -> str`¶

`get_sessions(patient_id: str, plan_id: str = '') -> list[dict]`¶

`add_outcome_measure(patient_id: str, instrument: str = '', baseline_score: float = 0.0, current_score: float = 0.0, target_score: float = 0.0, measured_date: str = '') -> str`¶

`get_outcome_measures(patient_id: str, instrument: str = '') -> list[dict]`¶

`update_outcome_score(measure_id: str, current_score: float, measured_date: str = '') -> None`¶

Functions¶

`assemble_review_text(output: Any) -> str`¶

Collect the reviewable free text from a JobAlliedHealthOutput (duck-typed).

`assess_allied_health_output(output: Any, qa_block: Any | None = None, generate: Any | None = None) -> GroundedRunResult`¶

Run grounded four-valued QA over an allied-health output.

MCP Tools¶

Operation	Source
`assess_patient`	allied_health_mcp
`plan_therapy`	allied_health_mcp
`track_progress`	allied_health_mcp
`document_treatment`	allied_health_mcp
`refer_specialist`	allied_health_mcp
`analyze_outcomes`	allied_health_mcp
`evaluate_intervention`	allied_health_mcp
`assess_mobility`	allied_health_mcp
`benchmark_recovery`	allied_health_mcp
`audit_documentation`	allied_health_mcp
`create_proposal`	allied_health_mcp
`review_deliverable`	allied_health_mcp
`delegate_task`	allied_health_mcp
`report_status`	allied_health_mcp
`request_feedback`	allied_health_mcp
`store_artifact`	allied_health_mcp
`retrieve_artifact`	allied_health_mcp
`list_artifacts`	allied_health_mcp
`search_artifacts`	allied_health_mcp
`archive`	allied_health_mcp
`plan_sprint`	allied_health_mcp
`reflect_on_outcome`	allied_health_mcp
`list_patterns`	allied_health_mcp
`get_capabilities`	allied_health_mcp
`info`	allied_health_mcp