Model Context Protocol (MCP)#

Client                          Server
  │                               │
  │── initialize ──────────────>  │  Exchange capabilities
  │<─ initialized ─────────────   │
  │                               │
  │── tools/list ──────────────>  │  Discover available tools
  │<─ {tools: [...]} ──────────   │
  │                               │
  │── tools/call ──────────────>  │  Execute tool
  │   {name, arguments}           │
  │<─ {content: [...]} ────────   │  Return result

"""
MCP server exposing clinical data tools for HMS integration.
Educational Example — Illustrative MCP server implementation.
Not intended for clinical decision making.

Prerequisites:
    pip install mcp  # Verify current package name at modelcontextprotocol.io
"""
from __future__ import annotations

import json
from typing import Any

try:
    import mcp.server.stdio
    import mcp.types as types
    from mcp.server import NotificationOptions, Server
    from mcp.server.models import InitializationOptions
    MCP_AVAILABLE = True
except ImportError:
    MCP_AVAILABLE = False
    print("MCP SDK not installed. See modelcontextprotocol.io for installation instructions.")


# ── Stub backend clients ──────────────────────────────────────────────────────

class FHIRClient:
    """Stub FHIR R4 client — production uses real FHIR R4 endpoint."""
    def get_patient(self, patient_id: str) -> dict:
        return {
            "resourceType": "Patient",
            "id": patient_id,
            "name": [{"family": "[SURNAME]", "given": ["[GIVEN]"]}],
            "gender": "unknown",
            "birthDate": "1970-01-01",
        }

    def get_conditions(self, patient_id: str) -> list[dict]:
        return [
            {"code": "G47.33", "display": "Obstructive sleep apnea (adult)"},
            {"code": "E11.9", "display": "Type 2 diabetes mellitus without complications"},
        ]

    def get_medications(self, patient_id: str) -> list[dict]:
        return [
            {"medication": "Metformin 500 mg", "status": "active"},
            {"medication": "Lisinopril 10 mg", "status": "active"},
        ]


class GuidelinesSearchClient:
    """Stub guidelines search client — production queries vector store."""
    def search(self, query: str, limit: int = 3) -> list[dict]:
        return [
            {
                "id": "guideline-001",
                "title": "Synthetic Guideline (Educational Example Only)",
                "content": "This is placeholder guideline content for illustration purposes.",
                "relevance_score": 0.85,
            }
        ]


if MCP_AVAILABLE:
    # ── Server setup ──────────────────────────────────────────────────────────

    fhir_client = FHIRClient()
    guidelines_client = GuidelinesSearchClient()

    server = Server("hms-clinical-mcp-server")

    # ── Tool definitions ──────────────────────────────────────────────────────

    @server.list_tools()
    async def handle_list_tools() -> list[types.Tool]:
        """Expose available tools to MCP clients."""
        return [
            types.Tool(
                name="get_patient_clinical_summary",
                description=(
                    "Retrieve a patient's clinical summary from the EHR system. "
                    "Returns demographics, active diagnoses, and current medications. "
                    "Use when clinical context about a specific patient is required."
                ),
                inputSchema={
                    "type": "object",
                    "properties": {
                        "patient_id": {
                            "type": "string",
                            "description": "Patient identifier (EHR system ID)",
                        }
                    },
                    "required": ["patient_id"],
                },
            ),
            types.Tool(
                name="search_clinical_guidelines",
                description=(
                    "Search the institutional clinical guidelines library using natural language. "
                    "Returns relevant guideline excerpts with source citations. "
                    "Use when clinical evidence or criteria are needed for a specific condition or procedure."
                ),
                inputSchema={
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "Natural language search query",
                        },
                        "max_results": {
                            "type": "integer",
                            "description": "Maximum results to return (1–10, default 3)",
                            "default": 3,
                            "minimum": 1,
                            "maximum": 10,
                        },
                    },
                    "required": ["query"],
                },
            ),
        ]

    # ── Tool handlers ─────────────────────────────────────────────────────────

    @server.call_tool()
    async def handle_call_tool(
        name: str, arguments: dict[str, Any] | None
    ) -> list[types.TextContent]:
        """Route tool calls to backend implementations."""
        args = arguments or {}

        if name == "get_patient_clinical_summary":
            patient_id = args.get("patient_id")
            if not patient_id:
                return [types.TextContent(
                    type="text",
                    text=json.dumps({"error": "patient_id is required"})
                )]

            patient = fhir_client.get_patient(patient_id)
            conditions = fhir_client.get_conditions(patient_id)
            medications = fhir_client.get_medications(patient_id)

            summary = {
                "patient_id": patient_id,
                "demographics": {
                    "gender": patient.get("gender"),
                    "birth_date": patient.get("birthDate"),
                },
                "active_diagnoses": [
                    {"code": c["code"], "display": c["display"]}
                    for c in conditions
                ],
                "current_medications": [
                    {"medication": m["medication"], "status": m["status"]}
                    for m in medications
                ],
                "disclaimer": "Educational Example — Not for clinical use",
            }

            return [types.TextContent(type="text", text=json.dumps(summary, indent=2))]

        elif name == "search_clinical_guidelines":
            query = args.get("query", "")
            max_results = min(args.get("max_results", 3), 10)

            results = guidelines_client.search(query, limit=max_results)

            response = {
                "query": query,
                "result_count": len(results),
                "results": results,
                "disclaimer": "Educational Example — Verify guideline accuracy before clinical use",
            }

            return [types.TextContent(type="text", text=json.dumps(response, indent=2))]

        else:
            return [types.TextContent(
                type="text",
                text=json.dumps({"error": f"Unknown tool: {name}"})
            )]

    # ── Resource definitions ──────────────────────────────────────────────────

    @server.list_resources()
    async def handle_list_resources() -> list[types.Resource]:
        """Expose static resources available to MCP clients."""
        return [
            types.Resource(
                uri="clinical://guidelines/index",
                name="Clinical Guidelines Index",
                description="Index of available clinical guideline categories in the institutional library",
                mimeType="application/json",
            ),
        ]

    @server.read_resource()
    async def handle_read_resource(uri: str) -> str:
        """Return resource content by URI."""
        if uri == "clinical://guidelines/index":
            index = {
                "categories": [
                    "Cardiology", "Endocrinology", "Neurology",
                    "Oncology", "Pulmonology", "Infectious Disease"
                ],
                "last_updated": "2026-01-15",
                "disclaimer": "Educational Example",
            }
            return json.dumps(index, indent=2)

        raise ValueError(f"Unknown resource URI: {uri}")

    # ── Server entry point ────────────────────────────────────────────────────

    async def run_server():
        """Run the MCP server over stdio transport."""
        async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
            await server.run(
                read_stream,
                write_stream,
                InitializationOptions(
                    server_name="hms-clinical-mcp-server",
                    server_version="1.0.0",
                    capabilities=server.get_capabilities(
                        notification_options=NotificationOptions(),
                        experimental_capabilities={},
                    ),
                ),
            )

"""
MCP client integration for a Claude-based application.
Educational Example — Illustrative client configuration.
Verify current MCP client API at modelcontextprotocol.io/docs
"""
from __future__ import annotations

import asyncio
import json
from typing import Any

try:
    from anthropic import Anthropic
    from mcp import ClientSession, StdioServerParameters
    from mcp.client.stdio import stdio_client
    MCP_CLIENT_AVAILABLE = True
except ImportError:
    MCP_CLIENT_AVAILABLE = False


async def run_agent_with_mcp_server(
    user_query: str,
    server_script_path: str,
    system_prompt: str = "",
) -> str:
    """
    Run a Claude agent that uses tools from a local MCP server.

    The agent:
    1. Connects to the MCP server via stdio
    2. Discovers available tools
    3. Runs the agent loop, routing tool calls to the MCP server
    """
    if not MCP_CLIENT_AVAILABLE:
        return "MCP client SDK not available — install required packages"

    anthropic_client = Anthropic()

    server_params = StdioServerParameters(
        command="python",
        args=[server_script_path],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Discover tools from the MCP server
            tools_response = await session.list_tools()

            # Convert MCP tool definitions to Anthropic API tool format
            anthropic_tools = [
                {
                    "name": tool.name,
                    "description": tool.description,
                    "input_schema": tool.inputSchema,
                }
                for tool in tools_response.tools
            ]

            messages = [{"role": "user", "content": user_query}]

            # Agent loop
            for _ in range(10):  # max_iterations circuit breaker
                response = anthropic_client.messages.create(
                    model="claude-opus-4-8",  # Verify current model ID at docs.anthropic.com
                    max_tokens=4096,
                    system=system_prompt,
                    tools=anthropic_tools,
                    messages=messages,
                )

                messages.append({"role": "assistant", "content": response.content})

                if response.stop_reason == "end_turn":
                    # Extract final text response
                    for block in response.content:
                        if hasattr(block, "text"):
                            return block.text
                    return "No text response produced"

                if response.stop_reason == "tool_use":
                    tool_results = []

                    for block in response.content:
                        if block.type != "tool_use":
                            continue

                        # Route tool call through MCP session
                        mcp_result = await session.call_tool(
                            name=block.name,
                            arguments=block.input,
                        )

                        # Convert MCP result to Anthropic tool result format
                        result_content = ""
                        if mcp_result.content:
                            result_content = mcp_result.content[0].text if hasattr(mcp_result.content[0], "text") else str(mcp_result.content[0])

                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": result_content,
                        })

                    messages.append({"role": "user", "content": tool_results})

            return "Max iterations reached without final response"


def run_mcp_agent_example():
    """Example entry point — illustrative use."""
    system = (
        "You are a clinical data assistant for a Reference Healthcare Organization. "
        "Use the available tools to retrieve patient information and clinical guidelines "
        "when needed to answer questions. "
        "Educational Example — Not for clinical decision making."
    )

    result = asyncio.run(run_agent_with_mcp_server(
        user_query="Retrieve the clinical summary for patient P-DEMO-001",
        server_script_path="examples/mcp/01-mcp-server-example.py",
        system_prompt=system,
    ))
    print(result)