Job Psychologist¶

job_psychologist — G6 Psychologist job agent.

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

Overview¶

Domain-specialist job agent for psychologists and counsellors. Supports client assessment, treatment plan development, session documentation, literature review, outcome evaluation, and psychometric analysis — all within a strictly human-review-flagged, safety-bounded JobAgentBlock framework designed as a clinical decision-support tool rather than an autonomous practitioner.

When to use:

• Generating structured assessment summaries from presenting concern narratives and test scores
• Drafting evidence-based treatment plans aligned to CBT, ACT, or DBT frameworks
• Producing structured session notes compliant with professional documentation standards
• Reviewing recent literature on a specific condition or intervention to inform practice

Example:

from mvp.job_psychologist import JobPsychologistBlock, JobPsychologistInput

block = JobPsychologistBlock()
result = block.infer(JobPsychologistInput(
    task="Summarise the intake assessment for a client presenting with generalised anxiety and draft a 12-week CBT treatment plan",
    context={"presenting_concern": "GAD", "GAD7_score": 14, "PHQ9_score": 8, "requires_human_review": True},
))
# result.ok → True; result.value → JobPsychologistOutput with result, artifacts

Works well with: job_framework, job_allied_health, job_education

Public API¶

PsychologistDecision¶

Validated op-classification decision for a clinical task.

PsychologistPlanner¶

Runtime-first facade with deterministic fallback (ACTION tier).

Constructor:

Methods:

classify(task: str, context: dict | None = None) -> PsychologistDecision¶

JobPsychologistBlock(JobAgentBlock)¶

G6 Psychologist job agent — crisis detection + MCP delegation with safety gate.

Methods:

infer(data)¶

Public entrypoint with an UNBYPASSABLE crisis precheck (Codex CRITICAL fix).

JobPsychologistInput(JobInput)¶

Input for the Psychologist job agent.

JobPsychologistOutput(JobOutput)¶

Output from the Psychologist job agent.

ClinicalRecordAdapter(ABC)¶

Abstract base class for clinical record storage backends.

Methods:

create_client(demographics: dict, presenting_concerns: str, referral_source: str = '', risk_level: str = 'low') -> str¶

Create a new client record. Returns client_id.

get_client(client_id: str) -> dict | None¶

Retrieve a client record by ID.

list_clients(status: str = '', limit: int = 50) -> list[dict]¶

List client records, optionally filtered by status.

create_assessment(client_id: str, instrument: str, raw_scores: dict, interpretation: str = '') -> str¶

Create an assessment record. Returns assessment_id.

get_assessments(client_id: str, instrument: str = '') -> list[dict]¶

Retrieve assessments for a client.

create_session(client_id: str, session_number: int, progress_notes: str = '', interventions: list | None = None) -> str¶

Create a session record. Returns session_id.

get_sessions(client_id: str) -> list[dict]¶

Retrieve sessions for a client.

create_treatment_plan(client_id: str, diagnosis_codes: list, goals: list, modality: str = 'individual') -> str¶

Create a treatment plan. Returns plan_id.

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

Retrieve treatment plans for a client.

create_outcome_record(client_id: str, measure: str, baseline: float, current: float, target: float, session_number: int = 0) -> str¶

Create an outcome tracking record. Returns tracking_id.

get_outcome_records(client_id: str, measure: str = '') -> list[dict]¶

Retrieve outcome records for a client.

InMemoryClinicalAdapter(ClinicalRecordAdapter)¶

In-memory implementation for testing and development.

Methods:

create_client(demographics: dict, presenting_concerns: str, referral_source: str = '', risk_level: str = 'low') -> str¶

get_client(client_id: str) -> dict | None¶

list_clients(status: str = '', limit: int = 50) -> list[dict]¶

create_assessment(client_id: str, instrument: str, raw_scores: dict, interpretation: str = '') -> str¶

get_assessments(client_id: str, instrument: str = '') -> list[dict]¶

create_session(client_id: str, session_number: int, progress_notes: str = '', interventions: list | None = None) -> str¶

get_sessions(client_id: str) -> list[dict]¶

create_treatment_plan(client_id: str, diagnosis_codes: list, goals: list, modality: str = 'individual') -> str¶

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

create_outcome_record(client_id: str, measure: str, baseline: float, current: float, target: float, session_number: int = 0) -> str¶

get_outcome_records(client_id: str, measure: str = '') -> list[dict]¶

JobPsychologistMCPBlock(AIBlock[MCPJobPsychologistInput, MCPJobPsychologistOutput, dict])¶

26-op MCP block for the Psychologist job agent.

Methods:

infer(data: MCPJobPsychologistInput) -> Result[MCPJobPsychologistOutput]¶

MCPJobPsychologistInput(BaseModel)¶

Input to JobPsychologistMCPBlock — 26-op dispatch.

MCPJobPsychologistOutput(BaseModel)¶

Output from JobPsychologistMCPBlock.

PsychologistStore(JobStore)¶

SQLite store for the Psychologist job agent.

Constructor:

Methods:

create_client(demographics: dict | None = None, presenting_concerns: str = '', referral_source: str = '', risk_level: str = 'low', status: str = 'active') -> str¶

get_client(client_id: str) -> dict | None¶

list_clients(status: str = '', limit: int = 50) -> list[dict]¶

update_client(client_id: str, **fields: Any) -> bool¶

create_assessment(client_id: str, instrument: str = '', raw_scores: dict | None = None, scaled_scores: dict | None = None, interpretation: str = '', administered_date: str = '') -> str¶

get_assessments(client_id: str, instrument: str = '', limit: int = 50) -> list[dict]¶

create_session(client_id: str, session_number: int = 1, session_type: str = 'individual', modality: str = 'in_person', interventions: list | None = None, progress_notes: str = '', duration_mins: int = 50, session_date: str = '') -> str¶

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

create_treatment_plan(client_id: str, diagnosis_codes: list | None = None, goals: list | None = None, interventions: list | None = None, modality: str = 'individual', status: str = 'active', review_date: str = '') -> str¶

get_treatment_plans(client_id: str, status: str = '', limit: int = 50) -> list[dict]¶

create_outcome_record(client_id: str, measure: str = '', baseline: float = 0.0, current: float = 0.0, target: float = 0.0, session_number: int = 0, measured_date: str = '') -> str¶

get_outcome_records(client_id: str, measure: str = '', limit: int = 100) -> list[dict]¶

Functions¶

summarize_job_psychologist_agentic_evidence(decisions: list[dict[str, Any]]) -> dict[str, Any]¶

Summarise runtime-vs-fallback psychologist decisions with path redaction.

agentic_planner_enabled(default_enabled: bool) -> bool¶

Decide whether the agentic psychologist planner should be used.

deterministic_classify(task: str, context: dict | None = None) -> PsychologistDecision¶

Demoted real keyword op-classifier -- the honest offline floor.

assemble_review_text(output: Any) -> str¶

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

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

Run grounded four-valued QA over a psychologist output.

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

Return compact metadata for the psychologist-applied vendored patterns.

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

Compact roll-up used by the list_patterns handler / skill catalog.

get_skill_catalog() -> PsychologistSkillCatalog¶

MCP Tools¶