mirrors/fhir-mcp-server
1
0
Fork 0
mirror of https://github.com/wso2/fhir-mcp-server.git synced 2025-12-07 04:18:25 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #3 from VimukthiRajapaksha/main

Browse Source 
Feature: Add docker support
This commit is contained in:
Nirmal Fernando
2025-07-03 10:21:02 +05:30
committed by GitHub
parent e94c8d4122 3692a7448a
commit b8bb620a28
9 changed files with 482 additions and 197 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
.choreo/component.yaml Normal file
View File

@@ -0,0 +1,40 @@
# Copyright (c) 2025, WSO2 LLC. (https://www.wso2.com/) All Rights Reserved.
#
# WSO2 LLC. licenses this file to you under the Apache License,
# Version 2.0 (the "License"); you may not use this file except
# in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# +required The configuration file schema version
schemaVersion: 1.2
# +optional Incoming connection details for the component
endpoints:
# +required Unique name for the endpoint.
# This name will be used when generating the managed API
- name: fhir_mcp_server
# +optional Display name for the endpoint.
displayName: FHIR MCP Server
# +required Service section has the user service endpoint details
service:
# +optional Base path of the API that gets exposed via the endpoint.
# This is mandatory if the endpoint type is set to REST or GraphQL.
basePath: /
# +required Numeric port value that gets exposed via the endpoint
port: 8000
# +required Type of traffic that the endpoint is accepting.
# Allowed values: REST, GraphQL, GRPC, TCP, UDP.
type: REST
# +optional Network level visibilities of the endpoint.
# Accepted values: Project|Organization|Public(Default).
networkVisibilities:
- Public

45
.dockerignore Normal file
View File

@@ -0,0 +1,45 @@
# Ignore Python cache and build artifacts
__pycache__/
*.pyc
*.pyo
*.pyd
*.py[co]
# Ignore virtual environments
.venv/
venv/
ENV/
# Ignore distribution/build directories
build/
dist/
wheels/
*.egg-info/
# Ignore environment files
.env
*.env
# Ignore IDE/editor config
.vscode/
.idea/
# Ignore OS files
.DS_Store
Thumbs.db
# Ignore test and coverage outputs
.coverage
htmlcov/
.tox/
# Ignore git and docker files
.git
.gitignore
.dockerignore
# Ignore logs
*.log
# Ignore documentation builds
docs/_build/

39
Dockerfile Normal file
View File

@@ -0,0 +1,39 @@
# ----------------------------------------------------------------------------------------
#
# Copyright (c) 2025, WSO2 LLC. (https://www.wso2.com). All Rights Reserved.
#
# This software is the property of WSO2 LLC. and its suppliers, if any.
# Dissemination of any information or reproduction of any material contained
# herein in any form is strictly forbidden, unless permitted by WSO2 expressly.
# You may not alter or remove any copyright or other notice from copies of this content.
#
# ----------------------------------------------------------------------------------------
FROM python:3.11-slim
# Install uv (for fast dependency management)
RUN pip install --upgrade pip
# Set workdir
WORKDIR /app
# Copy only requirements first for better caching
COPY requirements.txt ./
RUN pip install -r requirements.txt
# Copy the rest of the code
COPY . .
# Create a non-root user with UID 10001 and switch to it
RUN useradd -m -u 10001 appuser
USER 10001
# Expose default port
EXPOSE 8000
# Set environment variables (can be overridden at runtime)
ENV PYTHONUNBUFFERED=1
ENV PYTHONPATH=/app/src
# Default command to run the server (can be overridden)
CMD ["python", "-m", "fhir_mcp_server"]

8
pyproject.toml
View File

@@ -10,9 +10,9 @@ classifiers = [
"Operating System :: OS Independent",
]
dependencies = [
"mcp[cli]>=1.9.1",
"aiohttp>=3.12.13",
"fhirpy>=2.0.15",
"mcp[cli]==1.9.1",
"aiohttp==3.12.13",
"fhirpy==2.0.15",
]
description = "A Model Context Protocol (MCP) server that provides seamless, standardized access to Fast Healthcare Interoperability Resources (FHIR) data from any compatible FHIR server. Designed for easy integration with AI tools, developer workflows, and healthcare applications, it enables natural language and programmatic search, retrieval, and analysis of clinical data."
keywords = [
@@ -28,7 +28,7 @@ license-files = ["LICEN[CS]E*"]
name = "fhir-mcp-server"
readme = {file = "README.md", content-type = "text/markdown"}
requires-python = ">=3.12"
version = "1.0.1"
version = "0.1.11"
[build-system]
build-backend = "hatchling.build"

5
src/fhir_mcp_server/oauth/common.py
View File

@@ -128,7 +128,10 @@ def generate_code_challenge(code_verifier: str) -> str:
async def perform_token_flow(
url: str,
data: Dict[str, str],
headers: Dict[str, str] = {"Content-Type": "application/x-www-form-urlencoded"},
headers: Dict[str, str] = {
"Content-Type": "application/x-www-form-urlencoded",
"Accept": "application/json",
},
timeout: float = 30.0,
) -> OAuthToken:
try:

2
src/fhir_mcp_server/oauth/types.py
View File

@@ -41,7 +41,7 @@ class MCPOAuthConfigs(BaseOAuthConfigs):
class FHIROAuthConfigs(BaseOAuthConfigs):
base_url: str = "https://hapi.fhir.org/baseR5"
base_url: str = ""
timeout: int = 30 # in secs
access_token: str | None = None

499
src/fhir_mcp_server/server.py
View File

@@ -21,11 +21,11 @@ from fhir_mcp_server.utils import (
create_async_fhir_client,
get_bundle_entries,
get_default_headers,
get_operation_outcome_error,
get_operation_outcome,
get_operation_outcome_exception,
get_operation_outcome_required_error,
get_capability_statement,
trim_resource,
trim_resource_capabilities,
)
from fhir_mcp_server.oauth import (
handle_failed_authentication,
@@ -37,10 +37,11 @@ from fhir_mcp_server.oauth import (
)
from fhirpy import AsyncFHIRClient
from fhirpy.lib import AsyncFHIRResource
from fhirpy.base.exceptions import OperationOutcome
from fhirpy.base.exceptions import OperationOutcome, ResourceNotFound
from fhirpy.base.searchset import Raw
from typing import Dict, Any, Optional
from pydantic import AnyHttpUrl
from typing import Dict, Any, List
from typing_extensions import Annotated
from pydantic import AnyHttpUrl, Field
from starlette.requests import Request
from starlette.responses import RedirectResponse, Response, HTMLResponse
from mcp.server.auth.middleware.auth_context import get_access_token
@@ -198,30 +199,41 @@ def register_mcp_tools(mcp: FastMCP) -> None:
"""
logger.debug("Registering MCP tools.")
@mcp.tool()
async def get_capabilities(type: str) -> Dict[str, Any]:
"""
Retrieves metadata about a specified FHIR resource type, including its supported search parameters and custom operations.
This tool should be used at the start of any workflow where you need to discover what queries or operations are permitted
against that resource (e.g., before calling search, read, or create). Do not use this tool to fetch actual resources.
It only returns definitions and descriptions of capabilities, not resource instances. Because FHIR defines different search
parameters and operations per resource type, this tool ensures your subsequent calls use valid inputs.
Args:
type (str): The FHIR resource type name (e.g., "Patient", "Observation", "Encounter").
Must exactly match one of the core or profile-defined resource types supported by the server.
Returns:
Dict[str, Any]:
A dictionary containing:
- "type" (str): The requested resource type (if available) or empty.
- "searchParam" (Dict[str, str]): A map of FHIR search-parameter names. Each key is the parameter name
(e.g., "family", "_id", "_lastUpdated"), and each value is the FHIR-provided description of that parameter's meaning and usage constraints.
- "operation" (Dict[str, str]): A map of custom FHIR operation names to their descriptions.
Each key is the operation name (e.g., "$validate"), and each value explains the operation's purpose.
"""
@mcp.tool(
description=(
"Retrieves metadata about a specified FHIR resource type, including its supported search parameters and custom operations. "
"This tool MUST always be invoked before performing any resource operation (such as search, read, create, update, or delete) "
"to discover the valid searchParams and operations permitted for that resource type. "
"Do not use this tool to fetch actual resources."
)
)
async def get_capabilities(
type: Annotated[
str,
Field(
description=(
"The FHIR resource type name. Must exactly match one of the core or "
"profile-defined resource types as per the FHIR specification."
),
examples=["Patient", "Observation", "Encounter"],
),
],
) -> Annotated[
Dict[str, Any],
Field(
description=(
"A dictionary containing: "
"'type': The requested resource type (if supported by the system) or empty. "
"'searchParam': A mapping of FHIR search parameter names to their descriptions. Each key is a parameter name "
"(e.g., family, _id, _lastUpdated), and each value is a string describing the parameter's meaning and usage constraints. "
"'operation': A mapping of custom FHIR operation names to their descriptions. Each key is an operation name "
"(e.g., $validate), and each value is a string explaining the operation's purpose and usage. "
"'interaction': A list of supported interactions for the resource type (e.g., read, search-type, create). "
"'searchInclude': A list of supported _include parameters for the resource type, indicating which related resources can be included. "
"'searchRevInclude': A list of supported _revinclude parameters for the resource type, indicating which reverse-included resources can be included."
)
),
]:
try:
logger.debug(f"Invoked with resource_type='{type}'")
data: Dict[str, Any] = await get_capability_statement(
@@ -234,11 +246,18 @@ def register_mcp_tools(mcp: FastMCP) -> None:
)
return {
"type": resource.get("type"),
"searchParam": trim_resource(resource.get("searchParam", [])),
"operation": trim_resource(resource.get("operation", [])),
"searchParam": trim_resource_capabilities(
resource.get("searchParam", [])
),
"operation": trim_resource_capabilities(
resource.get("operation", [])
),
"interaction": resource.get("interaction", []),
"searchInclude": resource.get("searchInclude", []),
"searchRevInclude": resource.get("searchRevInclude", []),
}
logger.info(f"Resource type '{type}' not found in the CapabilityStatement.")
return await get_operation_outcome_error(
return await get_operation_outcome(
code="not-supported",
diagnostics=f"The interaction, operation, resource or profile {type} is not supported.",
)
@@ -249,27 +268,40 @@ def register_mcp_tools(mcp: FastMCP) -> None:
)
return await get_operation_outcome_exception()
@mcp.tool()
@mcp.tool(
description=(
"Executes a standard FHIR `search` interaction on a given resource type, returning a bundle or list of matching resources. "
"Use this when you need to query for multiple resources based on one or more search-parameters. "
"Do not use this tool for create, update, or delete operations, and be aware that large result sets may be paginated by the FHIR server."
)
)
async def search(
type: str, searchParam: Dict[str, str]
) -> list[AsyncFHIRResource] | Dict[str, Any]:
"""
Executes a standard FHIR "search" interaction on a given resource type, returning a bundle or list of matching resources.
Use this when you need to query for multiple resources based on one or more search-parameters.
Do not use this tool for create, update, or delete operations, and be aware that large result sets may be paginated by the FHIR server.
Args:
type (str): The FHIR resource type name (e.g., "MedicationRequest", "Condition", "Procedure").
Must exactly match one of the core or profile-defined resource types supported by the server.
searchParam (Dict[str, str]): A mapping of FHIR search parameter names to their desired values (e.g., {"family":"Smith","birthdate":"1970-01-01"}).
These parameters refine queries for operation-specific query qualifiers.
Only parameters exposed by `get_capabilities` for that resource type are valid.
Returns:
Dict[str, Any]: A dictionary containing the full FHIR resource instance matching the search criteria.
"""
type: Annotated[
str,
Field(
description="The FHIR resource type name. Must exactly match one of the resource types supported by the server",
examples=["MedicationRequest", "Condition", "Procedure"],
),
],
searchParam: Annotated[
Dict[str, str | List[str]],
Field(
description=(
"A mapping of FHIR search parameter names to their values. "
"Only include parameters supported for the resource type, as listed by `get_capabilities`."
),
examples=[
'{"family": "Smith"}',
'{"date": ["ge1970-01-01", "lt2000-01-01"]}',
],
),
],
) -> Annotated[
list[Dict[str, Any]] | Dict[str, Any],
Field(
description="A dictionary containing the full FHIR resource instance matching the search criteria."
),
]:
try:
logger.debug(f"Invoked with type='{type}' and searchParam={searchParam}")
if not type:
@@ -279,13 +311,19 @@ def register_mcp_tools(mcp: FastMCP) -> None:
return await get_operation_outcome_required_error("type")
client: AsyncFHIRClient = await get_async_fhir_client()
return await client.resources(type).search(Raw(**searchParam)).fetch()
async_resources: list[AsyncFHIRResource] = (
await client.resources(type).search(Raw(**searchParam)).fetch()
)
resources: list[Dict[str, Any]] = []
for async_resource in async_resources:
resources.append(async_resource.serialize())
return resources
except ValueError as ex:
logger.exception(
f"User does not have permission to perform FHIR '{type}' resource search operation. Caused by, ",
exc_info=ex,
)
return await get_operation_outcome_error(
return await get_operation_outcome(
code="forbidden",
diagnostics=f"The user does not have the rights to perform search operation.",
)
@@ -302,34 +340,53 @@ def register_mcp_tools(mcp: FastMCP) -> None:
)
return await get_operation_outcome_exception()
@mcp.tool()
@mcp.tool(
description=(
"Performs a FHIR `read` interaction to retrieve a single resource instance by its type and resource ID, "
"optionally refining the response with search parameters or custom operations. "
"Use it when you know the exact resource ID and require that one resource; do not use it for bulk queries. "
"If additional query-level parameters or operations are needed (e.g., _elements or $validate), include them in searchParam or operation."
)
)
async def read(
type: str,
id: str,
searchParam: Optional[Dict[str, str]] = None,
operation: Optional[str] = "",
) -> Dict[str, Any]:
"""
Performs a FHIR "read" interaction to retrieve a single resource instance by its type and resource ID,
optionally refining the response with search parameters or custom operations.
Use it when you know the exact resource ID and require that one resource; do not use it for bulk queries.
If additional query-level parameters or operations are needed (e.g., _elements or $validate), include them in searchParam or operation.
Args:
type (str): The FHIR resource type name (e.g., "DiagnosticReport", "AllergyIntolerance", "Immunization").
Must exactly match one of the core or profile-defined resource types supported by the server.
id (str): The logical ID of a specific FHIR resource instance.
searchParam (Dict[str, str]): A mapping of FHIR search parameter names to their desired values (e.g., {"device-name":"glucometer"}).
These parameters refine queries for operation-specific query qualifiers.
Only parameters exposed by `get_capabilities` for that resource type are valid.
operation (Optional[str]): The name of a custom FHIR operation or extended query defined for the resource (e.g., "$everything").
Must match one of the operation names returned by `get_capabilities`.
Returns:
Dict[str, Any]: A dictionary containing the single FHIR resource instance of the requested type and id.
"""
type: Annotated[
str,
Field(
description="The FHIR resource type name. Must exactly match one of the resource types supported by the server.",
examples=["DiagnosticReport", "AllergyIntolerance", "Immunization"],
),
],
id: Annotated[
str,
Field(description="The logical ID of a specific FHIR resource instance."),
],
searchParam: Annotated[
Dict[str, str | List[str]],
Field(
description=(
"A mapping of FHIR search parameter names to their desired values. "
"These parameters refine queries for operation-specific query qualifiers. "
"Only parameters exposed by `get_capabilities` for that resource type are valid."
),
examples=['{"device-name": "glucometer", "identifier": ["12345"]}'],
),
] = {},
operation: Annotated[
str,
Field(
description=(
"The name of a custom FHIR operation or extended query defined for the resource "
"must match one of the operation names returned by `get_capabilities`."
),
examples=["$everything"],
),
] = "",
) -> Annotated[
Dict[str, Any],
Field(
description="A dictionary containing the single FHIR resource instance of the requested type and id."
),
]:
try:
logger.debug(
f"Invoked with type='{type}', id={id}, searchParam={searchParam}, and operation={operation}"
@@ -346,12 +403,21 @@ def register_mcp_tools(mcp: FastMCP) -> None:
)
return await get_bundle_entries(bundle=bundle)
except ResourceNotFound as ex:
logger.error(
f"Resource of type '{type}' with id '{id}' not found. Caused by, ",
exc_info=ex,
)
return await get_operation_outcome(
code="not-found",
diagnostics=f"The resource of type '{type}' with id '{id}' was not found.",
)
except ValueError as ex:
logger.exception(
f"User does not have permission to perform FHIR '{type}' resource read operation. Caused by, ",
exc_info=ex,
)
return await get_operation_outcome_error(
return await get_operation_outcome(
code="forbidden",
diagnostics=f"The user does not have the rights to perform read operation.",
)
@@ -368,35 +434,61 @@ def register_mcp_tools(mcp: FastMCP) -> None:
)
return await get_operation_outcome_exception()
@mcp.tool()
@mcp.tool(
description=(
"Executes a FHIR `create` interaction to persist a new resource of the specified type. "
"It is required to supply the full resource payload in JSON form. "
"Use this tool when you need to add new data (e.g., a new Patient or Observation). "
"Note that servers may reject resources that violate profiles or mandatory bindings."
)
)
async def create(
type: str,
payload: Dict[str, Any],
searchParam: Optional[Dict[str, str]] = None,
operation: Optional[str] = "",
) -> Dict[str, Any]:
"""
Executes a FHIR "create" interaction to persist a new resource of the specified type. It is required to supply the full resource payload in JSON form.
Use this tool when you need to add new data (e.g., a new Patient or Observation). Do not call it to update existing resources; for updates, use patch.
Note that servers may reject resources that violate profiles or mandatory bindings.
Args:
type (str): The FHIR resource type name (e.g., "Device", "CarePlan", "Goal").
Must exactly match one of the core or profile-defined resource types supported by the server.
payload (Dict[str, str]): A JSON object representing the full FHIR resource body to be created.
It must include all required elements of the resource's profile.
searchParam (Dict[str, str]): A mapping of FHIR search parameter names to their desired values (e.g., {"address-city":"Boston"}).
These parameters refine queries for operation-specific query qualifiers.
Only parameters exposed by `get_capabilities` for that resource type are valid.
operation (Optional[str]): The name of a custom FHIR operation or extended query defined for the resource (e.g., "$evaluate").
Must match one of the operation names returned by `get_capabilities`.
Returns:
Dict[str, Any]: A dictionary containing the newly created FHIR resource, including server-assigned fields (id, meta.versionId, meta.lastUpdated,
and any server-added extensions). Reflects exactly what was persisted.
"""
type: Annotated[
str,
Field(
description="The FHIR resource type name. Must exactly match one of the resource types supported by the server.",
examples=["Device", "CarePlan", "Goal"],
),
],
payload: Annotated[
Dict[str, Any],
Field(
description=(
"A JSON object representing the full FHIR resource body to be created. "
"It must include all required elements of the resource's profile."
)
),
],
searchParam: Annotated[
Dict[str, str | List[str]],
Field(
description=(
"A mapping of FHIR search parameter names to their desired values. "
"These parameters refine queries for operation-specific query qualifiers. "
"Only parameters exposed by `get_capabilities` for that resource type are valid."
),
examples=['{"address-city": "Boston", "address-state": ["NY"]}'],
),
] = {},
operation: Annotated[
str,
Field(
description=(
"The name of a custom FHIR operation or extended query defined for the resource"
"Must match one of the operation names returned by `get_capabilities`."
),
examples=["$evaluate"],
),
] = "",
) -> Annotated[
Dict[str, Any],
Field(
description=(
"A dictionary containing the newly created FHIR resource, including server-assigned fields "
"(id, meta.versionId, meta.lastUpdated, and any server-added extensions). Reflects exactly what was persisted."
)
),
]:
try:
logger.debug(
f"Invoked with type='{type}', payload={payload}, searchParam={searchParam}, and operation={operation}"
@@ -418,7 +510,7 @@ def register_mcp_tools(mcp: FastMCP) -> None:
f"User does not have permission to perform FHIR '{type}' resource create operation. Caused by, ",
exc_info=ex,
)
return await get_operation_outcome_error(
return await get_operation_outcome(
code="forbidden",
diagnostics=f"The user does not have the rights to perform create operation.",
)
@@ -435,38 +527,63 @@ def register_mcp_tools(mcp: FastMCP) -> None:
)
return await get_operation_outcome_exception()
@mcp.tool()
@mcp.tool(
description=(
"Performs a FHIR `update` interaction by replacing an existing resource instance's content with the provided payload. "
"Use it when you need to overwrite a resource's data in its entirety, such as correcting or completing a record, "
"and you already know the resource's logical id. "
"Optionally, you can include searchParam for conditional updates (e.g., only update if the resource matches certain criteria) "
"or specify a custom operation (e.g., `$validate` to run validation before updating) "
"The tool returns the updated resource or an OperationOutcome detailing any errors."
)
)
async def update(
type: str,
id: str,
payload: Dict[str, Any],
searchParam: Optional[Dict[str, str]] = None,
operation: Optional[str] = "",
) -> Dict[str, Any]:
"""
Performs a FHIR "update" interaction by replacing an existing resource instance's content with the provided payload.
Use it when you need to overwrite a resource's data in its entirety, such as correcting or completing a record, and you already know the resource's logical id.
Optionally, you can include searchParam for conditional updates (e.g., only update if the resource matches certain criteria) or specify a
custom operation (e.g., "$validate" to run validation before updating). The tool returns the updated resource or an OperationOutcome detailing any errors.
Args:
type (str): The FHIR resource type name (e.g., "Location", "Organization", "Coverage").
id (str): The logical ID of a specific FHIR resource instance.
Must exactly match one of the core or profile-defined resource types supported by the server.
payload (Dict[str, Any]): The complete JSON representation of the FHIR resource, containing all required elements and any optional data.
Servers replace the existing resource with this exact content, so the payload must include all mandatory fields defined by the resource's profile
and any previous data you wish to preserve.
searchParam (Dict[str, str]): A mapping of FHIR search parameter names to their desired values (e.g., {"patient":"Patient/54321","relationship":"father"}).
These parameters refine queries for operation-specific query qualifiers.
Only parameters exposed by `get_capabilities` for that resource type are valid.
operation (Optional[str]): The name of a custom FHIR operation or extended query defined for the resource (e.g., "$lastn").
Must match one of the operation names returned by `get_capabilities`.
Returns:
Dict[str, Any]: A dictionary containing the updated FHIR resource after applying the JSON Patch operations..
"""
type: Annotated[
str,
Field(
description="The FHIR resource type name. Must exactly match one of the resource types supported by the server.",
examples=["Location", "Organization", "Coverage"],
),
],
id: Annotated[
str,
Field(description="The logical ID of a specific FHIR resource instance."),
],
payload: Annotated[
Dict[str, Any],
Field(
description=(
"The complete JSON representation of the FHIR resource, containing all required elements and any optional data. "
"Servers replace the existing resource with this exact content, so the payload must include all mandatory fields "
"defined by the resource's profile and any previous data you wish to preserve."
)
),
],
searchParam: Annotated[
Dict[str, str | List[str]],
Field(
description=(
"A mapping of FHIR search parameter names to their desired values. "
"These parameters refine queries for operation-specific query qualifiers. "
"Only parameters exposed by `get_capabilities` for that resource type are valid. "
),
examples=['{"patient":"Patient/54321","relationship":["father"]}'],
),
] = {},
operation: Annotated[
str,
Field(
description=(
"The name of a custom FHIR operation or extended query defined for the resource"
"Must match one of the operation names returned by `get_capabilities`."
),
examples=["$lastn"],
),
] = "",
) -> Annotated[
Dict[str, Any],
Field(description="A dictionary containing the updated FHIR resource"),
]:
try:
logger.debug(
f"Invoked with type='{type}', id={id}, payload={payload}, searchParam={searchParam}, and operation={operation}"
@@ -481,7 +598,7 @@ def register_mcp_tools(mcp: FastMCP) -> None:
bundle: dict = await client.resource(resource_type=type, id=id).execute(
operation=operation or "",
method="PUT",
data={id: id, **payload},
data={**payload, "id": id},
params=searchParam,
)
return await get_bundle_entries(bundle=bundle)
@@ -490,7 +607,7 @@ def register_mcp_tools(mcp: FastMCP) -> None:
f"User does not have permission to perform FHIR '{type}' resource update operation. Caused by, ",
exc_info=ex,
)
return await get_operation_outcome_error(
return await get_operation_outcome(
code="forbidden",
diagnostics=f"The user does not have the rights to perform update operation.",
)
@@ -507,36 +624,55 @@ def register_mcp_tools(mcp: FastMCP) -> None:
)
return await get_operation_outcome_exception()
@mcp.tool()
@mcp.tool(
description=(
"Execute a FHIR `delete` interaction on a specific resource instance. "
"Use this tool when you need to remove a single resource identified by its logical ID or optionally filtered by search parameters. "
"The optional `id` parameter must match an existing resource instance when present. "
"If you include `searchParam`, the server will perform a conditional delete, deleting the resource only if it matches the given criteria. "
"If you supply `operation`, it will execute the named FHIR operation (e.g., `$expunge`) on the resource. "
"This tool returns a FHIR `OperationOutcome` describing success or failure of the deletion."
)
)
async def delete(
type: str,
id: Optional[str] = "",
searchParam: Optional[Dict[str, str]] = None,
operation: Optional[str] = "",
) -> Dict[str, Any]:
"""
Execute a FHIR "delete" interaction on a specific resource instance.
Use this tool when you need to remove a single resource identified by its logical ID or optionally filtered by search parameters.
The optional `id` parameter must match an existing resource instance when present. If you include `searchParam`,
the server will perform a conditional delete, deleting the resource only if it matches the given criteria. If you supply `operation`,
it will execute the named FHIR operation (e.g., `$expunge`) on the resource. Do not use this tool for bulk deletes across multiple
This tool returns a FHIR `OperationOutcome` describing success or failure of the deletion.
Args:
type (str): The FHIR resource type name (e.g., "ServiceRequest", "Appointment", "HealthcareService").
id (str): The logical ID of a specific FHIR resource instance.
Must exactly match one of the core or profile-defined resource types supported by the server.
searchParam (Dict[str, str]): A mapping of FHIR search parameter names to their desired values (e.g., {"category":"laboratory","issued:"2025-05-01"}).
These parameters refine queries for operation-specific query qualifiers.
Only parameters exposed by `get_capabilities` for that resource type are valid.
operation (Optional[str]): The name of a custom FHIR operation or extended query defined for the resource (e.g., "$expand").
Must match one of the operation names returned by `get_capabilities`.
Returns:
Dict[str, Any]: A dictionary containing the confirmation of deletion or details on why deletion failed.
"""
type: Annotated[
str,
Field(
description="The FHIR resource type name. Must exactly match one of the resource types supported by the server.",
examples=["ServiceRequest", "Appointment", "HealthcareService"],
),
],
id: Annotated[
str,
Field(description="The logical ID of a specific FHIR resource instance."),
] = "",
searchParam: Annotated[
Dict[str, str | List[str]],
Field(
description=(
"A mapping of FHIR search parameter names to their desired values. "
"These parameters refine queries for operation-specific query qualifiers. "
"Only parameters exposed by `get_capabilities` for that resource type are valid. "
),
examples=['{"category": "laboratory", "status": ["active"]}'],
),
] = {},
operation: Annotated[
str,
Field(
description=(
"The name of a custom FHIR operation or extended query defined for the resource"
"Must match one of the operation names returned by `get_capabilities`."
),
examples=["$expand"],
),
] = "",
) -> Annotated[
Dict[str, Any],
Field(
description="A dictionary containing the confirmation of deletion or details on why deletion failed."
),
]:
try:
logger.debug(
f"Invoked with type='{type}', id={id}, searchParam={searchParam}, and operation={operation}"
@@ -546,18 +682,29 @@ def register_mcp_tools(mcp: FastMCP) -> None:
"Unable to perform delete operation: 'type' is a mandatory field."
)
return await get_operation_outcome_required_error("type")
if not id and not searchParam:
logger.error(
"Unable to perform delete operation: 'id' or 'searchParam' is required."
)
return await get_operation_outcome_required_error("id")
client: AsyncFHIRClient = await get_async_fhir_client()
bundle: dict = await client.resource(resource_type=type, id=id).execute(
bundle = await client.resource(resource_type=type, id=id).execute(
operation=operation or "", method="DELETE", params=searchParam
)
return await get_bundle_entries(bundle=bundle)
if isinstance(bundle, Dict):
return await get_bundle_entries(bundle=bundle)
return await get_operation_outcome(
severity="information",
code="SUCCESSFUL_DELETE",
diagnostics="Successfully deleted resource(s).",
)
except ValueError as ex:
logger.exception(
f"User does not have permission to perform FHIR '{type}' resource delete operation. Caused by, ",
exc_info=ex,
)
return await get_operation_outcome_error(
return await get_operation_outcome(
code="forbidden",
diagnostics=f"The user does not have the rights to perform delete operation.",
)

33
src/fhir_mcp_server/utils.py
View File

@@ -47,7 +47,7 @@ async def create_async_fhir_client(
async def get_bundle_entries(bundle: Dict[str, Any]) -> Dict[str, Any]:
if "entry" in bundle and isinstance(bundle["entry"], list):
if bundle and "entry" in bundle and isinstance(bundle["entry"], list):
logger.debug(f"found {len(bundle['entry'])} entries for type '{type}'")
return {
"entry": [
@@ -59,35 +59,46 @@ async def get_bundle_entries(bundle: Dict[str, Any]) -> Dict[str, Any]:
return bundle
def trim_resource(operations: List[Dict[str, Any]]) -> List[Dict[str, Optional[str]]]:
logger.debug(f"trim_resource called with {len(operations)} operations.")
def trim_resource_capabilities(
capabilities: List[Dict[str, Any]],
) -> List[Dict[str, Optional[str]]]:
logger.debug(
f"trim_resource_capabilities called with {len(capabilities)} capabilities."
)
trimmed = [
{"name": operation.get("name"), "documentation": operation.get("documentation")}
for operation in operations
if "name" in operation or "documentation" in operation
{
"name": capability.get("name"),
"documentation": capability.get("documentation"),
}
for capability in capabilities
if "name" in capability or "documentation" in capability
]
logger.debug(f"trim_resource returning {len(trimmed)} trimmed operations.")
logger.debug(
f"trim_resource_capabilities returning {len(trimmed)} trimmed capabilities."
)
return trimmed
async def get_operation_outcome_exception() -> dict:
return await get_operation_outcome_error(
return await get_operation_outcome(
code="exception", diagnostics="An unexpected internal error has occurred."
)
async def get_operation_outcome_required_error(element: str = "") -> dict:
return await get_operation_outcome_error(
return await get_operation_outcome(
code="required", diagnostics=f"A required element {element} is missing."
)
async def get_operation_outcome_error(code: str, diagnostics: str) -> dict:
async def get_operation_outcome(
code: str, diagnostics: str, severity: str = "error"
) -> dict:
return {
"resourceType": "OperationOutcome",
"issue": [
{
"severity": "error",
"severity": severity,
"code": code,
"diagnostics": diagnostics,
}

8
uv.lock generated
View File

@@ -173,7 +173,7 @@ wheels = [
[[package]]
name = "fhir-mcp-server"
version = "1.0.2"
version = "0.1.11"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
@@ -183,9 +183,9 @@ dependencies = [
[package.metadata]
requires-dist = [
{ name = "aiohttp", specifier = ">=3.12.13" },
{ name = "fhirpy", specifier = ">=2.0.15" },
{ name = "mcp", extras = ["cli"], specifier = ">=1.9.1" },
{ name = "aiohttp", specifier = "==3.12.13" },
{ name = "fhirpy", specifier = "==2.0.15" },
{ name = "mcp", extras = ["cli"], specifier = "==1.9.1" },
]
[[package]]