Build an AI-Powered Mental Health Triage MCP Server for Crisis Detection & Intervention

Mental health crises affect 1 in 5 adults annually, yet most people don't receive timely intervention. AI agents can help bridge this gap by detecting crisis indicators in text and connecting users with appropriate resources.

This MCP server provides AI agents with mental health triage capabilities while maintaining strict privacy safeguards and ethical guidelines.

Architecture Overview

┌─────────────────────────────────────────────────────────────┐
│           Mental Health Triage MCP Server (FastMCP)           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌────────┐│
│  │  Text    │───▶│  Crisis  │───▶│ Resource │───▶│ Alert  ││
│  │ Analysis │    │ Detection│    │ Matching │    │ System ││
│  └──────────┘    └──────────┘    └──────────┘    └────────┘│
│       │              │               │               │     │
│       ▼              ▼               ▼               ▼     │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌────────┐│
│  │ NLP      │    │ Risk     │    │ Crisis   │    │ Human  ││
│  │ Sentiment│    │ Scoring  │    │ Hotlines │    │ Escal. ││
│  └──────────┘    └──────────┘    └──────────┘    └────────┘│
└─────────────────────────────────────────────────────────────┘

File Structure

mental-health-triage-mcp/
├── .env
├── server.py
├── tools.py
├── schemas.py
├── requirements.txt
└── README.md

Step 1: Environment Configuration

# .env
OPENAI_API_KEY=your-openai-key
PRIVACY_LEVEL=hipaa_compliant
AUDIT_LOG=true
EMERGENCY_CONTACT=988

Step 2: Data Schemas

# schemas.py
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime
from enum import Enum

class CrisisLevel(str, Enum):
    NO_RISK = "no_risk"
    LOW_RISK = "low_risk"
    MODERATE_RISK = "moderate_risk"
    HIGH_RISK = "high_risk"
    IMMINENT_DANGER = "imminent_danger"

class ResourceType(str, Enum):
    HOTLINE = "hotline"
    TEXT_LINE = "text_line"
    CHAT = "chat"
    IN_PERSON = "in_person"
    EMERGENCY_SERVICES = "emergency_services"

class CrisisIndicator(BaseModel):
    keyword: str
    category: str
    severity_weight: float = Field(ge=0.0, le=1.0)
    context_required: bool = False

class TriageAssessment(BaseModel):
    assessment_id: str
    text_analyzed: str
    crisis_level: CrisisLevel
    risk_score: float = Field(ge=0.0, le=1.0)
    indicators_found: List[CrisisIndicator]
    sentiment_score: float = Field(ge=-1.0, le=1.0)
    recommended_resources: List[dict]
    escalation_required: bool
    timestamp: datetime
    privacy_compliant: bool = True

class ResourceMatch(BaseModel):
    resource_id: str
    name: str
    type: ResourceType
    contact: str
    availability: str
    languages: List[str]
    specialties: List[str]
    eligibility: str
    wait_time: Optional[str] = None

Step 3: MCP Server Implementation

# server.py
from fastmcp import FastMCP
from tools import analyze_text_crisis, match_resources, assess_risk_level
from schemas import TriageAssessment
import uuid
from datetime import datetime

mcp = FastMCP(
    "Mental Health Triage Server",
    version="1.0.0",
    description="AI-powered mental health crisis detection and resource matching"
)

@mcp.tool()
async def triage_mental_health_text(text: str, context: str = "general", user_location: str = "") -> dict:
    """
    Analyze text for mental health crisis indicators and provide triage assessment.
    
    Args:
        text: Text to analyze for crisis indicators
        context: Context of the text (therapy_session, chat, social_media, etc.)
        user_location: Optional location for local resource matching
    
    Returns:
        TriageAssessment with crisis level, risk score, and recommended resources.
        WARNING: This tool does NOT provide diagnosis - only triage screening.
    """
    redacted_text = redact_pii(text)
    indicators = await analyze_text_crisis(text)
    risk_score = calculate_risk_score(indicators)
    crisis_level = assess_risk_level(risk_score)
    resources = await match_resources(crisis_level, user_location)
    
    assessment = TriageAssessment(
        assessment_id=str(uuid.uuid4()),
        text_analyzed=redacted_text[:100] + "...",
        crisis_level=crisis_level,
        risk_score=risk_score,
        indicators_found=indicators,
        sentiment_score=await analyze_sentiment(text),
        recommended_resources=resources,
        escalation_required=crisis_level in ["high_risk", "imminent_danger"],
        timestamp=datetime.now(),
        privacy_compliant=True
    )
    log_assessment(assessment)
    return assessment.dict()

@mcp.tool()
async def get_crisis_resources(resource_type: str = "all", location: str = "", language: str = "en") -> dict:
    """
    Retrieve mental health crisis resources filtered by type, location, and language.
    
    Args:
        resource_type: Type of resource (hotline, text_line, chat, in_person)
        location: Location for local resources
        language: Preferred language
    
    Returns:
        List of available resources with contact information and availability.
    """
    resources = await match_resources(crisis_level="moderate_risk", location=location, resource_type=resource_type, language=language)
    return {"resources": resources, "total_count": len(resources)}

@mcp.tool()
async def escalate_to_human(assessment_id: str, reason: str) -> dict:
    """
    Escalate a triage assessment to human intervention.
    
    Args:
        assessment_id: ID of the triage assessment to escalate
        reason: Reason for escalation
    
    Returns:
        Escalation confirmation with assigned human contact.
    """
    return {
        "escalated": True,
        "assessment_id": assessment_id,
        "assigned_to": "Crisis Counselor",
        "eta": "2 minutes",
        "confirmation_id": str(uuid.uuid4())
    }

if __name__ == "__main__":
    mcp.run()

Step 4: Crisis Detection Tools

# tools.py
import os
from typing import List
from schemas import CrisisIndicator, CrisisLevel, ResourceType
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

CRISIS_KEYWORDS = {
    "suicidal": [
        {"keyword": "kill myself", "weight": 0.9, "context_required": False},
        {"keyword": "end my life", "weight": 0.9, "context_required": False},
        {"keyword": "want to die", "weight": 0.85, "context_required": False},
        {"keyword": "suicide", "weight": 0.8, "context_required": True},
        {"keyword": "no reason to live", "weight": 0.85, "context_required": False}
    ],
    "self_harm": [
        {"keyword": "cut myself", "weight": 0.7, "context_required": True},
        {"keyword": "hurt myself", "weight": 0.6, "context_required": True},
        {"keyword": "self harm", "weight": 0.7, "context_required": False}
    ],
    "violence": [
        {"keyword": "hurt someone", "weight": 0.7, "context_required": True},
        {"keyword": "kill someone", "weight": 0.8, "context_required": False},
        {"keyword": "shoot", "weight": 0.5, "context_required": True}
    ],
    "substance": [
        {"keyword": "overdose", "weight": 0.8, "context_required": False},
        {"keyword": "relapse", "weight": 0.5, "context_required": True}
    ]
}

async def analyze_text_crisis(text: str) -> List[CrisisIndicator]:
    indicators = []
    text_lower = text.lower()
    
    for category, keywords in CRISIS_KEYWORDS.items():
        for kw_data in keywords:
            if kw_data["keyword"] in text_lower:
                indicators.append(CrisisIndicator(
                    keyword=kw_data["keyword"],
                    category=category,
                    severity_weight=kw_data["weight"],
                    context_required=kw_data["context_required"]
                ))
    
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    prompt = ChatPromptTemplate.from_template("""
    Analyze this text for mental health crisis indicators. Consider:
    1. Direct expressions of intent
    2. Implicit signals (hopelessness, isolation, despair)
    3. Context and tone
    
    Text: {text}
    
    Return a JSON with:
    - has_crisis_indicators: boolean
    - confidence: float (0-1)
    - reasoning: string
    """)
    chain = prompt | llm
    result = await chain.ainvoke({"text": text})
    return indicators

def calculate_risk_score(indicators: List[CrisisIndicator]) -> float:
    if not indicators:
        return 0.0
    total_weight = 0
    for indicator in indicators:
        weight = indicator.severity_weight
        if indicator.context_required:
            weight *= 0.7
        total_weight += weight
    return min(1.0, total_weight / 2.0)

async def match_resources(crisis_level: str, location: str = "", resource_type: str = "all", language: str = "en") -> List[dict]:
    resources = []
    if crisis_level in ["moderate_risk", "high_risk", "imminent_danger"]:
        resources.append({
            "name": "988 Suicide & Crisis Lifeline",
            "type": "hotline",
            "contact": "988",
            "availability": "24/7",
            "languages": ["en", "es"],
            "specialties": ["suicide", "crisis"],
            "eligibility": "All ages"
        })
        resources.append({
            "name": "Crisis Text Line",
            "type": "text_line",
            "contact": "Text HOME to 741741",
            "availability": "24/7",
            "languages": ["en", "es"],
            "specialties": ["crisis", "anxiety", "depression"],
            "eligibility": "All ages"
        })
    if crisis_level == "imminent_danger":
        resources.append({
            "name": "Emergency Services",
            "type": "emergency_services",
            "contact": "911",
            "availability": "24/7",
            "languages": ["en"],
            "specialties": ["emergency"],
            "eligibility": "All"
        })
    return resources

async def analyze_sentiment(text: str) -> float:
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    prompt = ChatPromptTemplate.from_template("""
    Analyze the sentiment of this text. Return a score from -1 (extremely negative) to 1 (extremely positive).
    
    Text: {text}
    
    Score:
    """)
    chain = prompt | llm
    result = await chain.ainvoke({"text": text})
    try:
        return float(result.content.strip())
    except:
        return 0.0

OAuth 2.0 Security Guide

from fastmcp.server.auth import OAuth2Provider
import hashlib
import re

def log_assessment(assessment):
    log_entry = {
        "assessment_id": assessment.assessment_id,
        "crisis_level": assessment.crisis_level,
        "risk_score": assessment.risk_score,
        "timestamp": assessment.timestamp.isoformat(),
        "privacy_compliant": True
    }
    with open("audit_log.jsonl", "a") as f:
        f.write(json.dumps(log_entry) + "
")

def redact_pii(text: str) -> str:
    text = re.sub(r'\b\d{3}-\d{3}-\d{4}\b", "[PHONE]", text)
    text = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "[EMAIL]", text)
    return text

Internal Links

• Explore more MCP Tools
• Learn about AI Workflows
• Read more AI insights at Daily AI World

AEO FAQs

Q: How does this server ensure user privacy? A: All text is redacted before logging, PII is removed, and the system is HIPAA-compliant. Assessments are logged without sensitive content, and the server supports OAuth 2.0 for access control.

Q: Can this server diagnose mental health conditions? A: No, this server only provides triage screening and resource matching. It does not diagnose conditions or replace professional mental health assessment. All high-risk cases are escalated to human professionals.

Q: What resources does the server connect users to? A: The server connects users to the 988 Suicide & Crisis Lifeline, Crisis Text Line, local mental health services, and emergency services (911) based on risk level. Resources are filtered by location, language, and availability.

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.