Source code for tooluniverse.icd_tool
"""ICD (International Classification of Diseases) API Tools.
WHO ICD-11 API for disease classification and coding.
Official API: https://icd.who.int/icdapi
Requires ICD_CLIENT_ID / ICD_CLIENT_SECRET, as environment variables or as
request credentials.
"""
import time
from urllib.parse import quote as url_quote
import requests
from typing import Any, Dict, Optional
from .base_tool import BaseTool
from .credentials import has_credential_context
from .tool_registry import register_tool
ICD_API_BASE = "https://id.who.int/icd"
ICD_API_AUTH = "https://icdaccessmanagement.who.int/connect/token"
[docs]
@register_tool("ICDTool")
class ICDTool(BaseTool):
"""WHO ICD-11 API tool for disease classification and coding."""
[docs]
def __init__(self, tool_config):
super().__init__(tool_config)
fields = tool_config.get("fields", {})
self.endpoint_template = fields.get("endpoint", "")
self.linearization = fields.get("linearization", "mms")
self.use_browser = fields.get("use_browser", False)
self._access_token: Optional[str] = None
self._token_expiry: float = 0
self._auth_error: Optional[str] = None
_REGISTER = (
"Set ICD_CLIENT_ID and ICD_CLIENT_SECRET as environment variables or as "
"request credentials. Register at: https://icd.who.int/icdapi"
)
[docs]
def _get_access_token(self) -> Optional[str]:
"""Get OAuth2 access token for ICD API (cached until near expiry).
Records why it failed in ``_auth_error``: a caller who has set both
variables and had them rejected must not be told to set them again.
"""
if (
not has_credential_context()
and self._access_token
and time.time() < self._token_expiry
):
return self._access_token
self._auth_error = None
client_id = self.credential("ICD_CLIENT_ID")
client_secret = self.credential("ICD_CLIENT_SECRET")
if not client_id or not client_secret:
missing = " and ".join(
name
for name, value in (
("ICD_CLIENT_ID", client_id),
("ICD_CLIENT_SECRET", client_secret),
)
if not value
)
self._auth_error = (
f"ICD API authentication required: {missing} is not set. "
f"{self._REGISTER}"
)
return None
try:
payload = {
"client_id": client_id,
"client_secret": client_secret,
"scope": "icdapi_access",
"grant_type": "client_credentials",
}
resp = requests.post(ICD_API_AUTH, data=payload, timeout=30)
if resp.status_code in (400, 401, 403):
# WHO answers a wrong id or secret with 400 invalid_client. That
# is a different problem from an unset variable and has a
# different fix, so it gets its own message.
detail = ""
try:
detail = resp.json().get("error") or ""
except ValueError:
detail = (resp.text or "").strip()[:100]
self._auth_error = (
"The ICD API rejected these ICD_CLIENT_ID / "
"ICD_CLIENT_SECRET credentials"
+ (f" ({detail})" if detail else "")
+ ". Check them, or register new ones at "
"https://icd.who.int/icdapi"
)
return None
resp.raise_for_status()
token_data = resp.json()
access_token = token_data.get("access_token")
if not access_token:
self._auth_error = (
"The ICD API returned no access token for these "
"credentials. " + self._REGISTER
)
return None
if not has_credential_context():
self._access_token = access_token
expires_in = token_data.get("expires_in", 3600)
self._token_expiry = time.time() + (expires_in - 60)
return access_token
except Exception as e:
self._auth_error = (
"Could not reach the ICD authentication service "
f"({ICD_API_AUTH}): {type(e).__name__}: {str(e)[:150]}"
)
return None
_PLACEHOLDER_KEYS = {
"{linearization}": "linearization",
"{entity_id}": "entity_id",
"{code}": "code",
}
[docs]
def _build_url(self, arguments: Dict[str, Any]) -> str:
"""Build the full URL from endpoint template and arguments."""
endpoint = self.endpoint_template
for placeholder, arg_key in self._PLACEHOLDER_KEYS.items():
if placeholder in endpoint:
default = self.linearization if arg_key == "linearization" else ""
endpoint = endpoint.replace(
placeholder, arguments.get(arg_key, default)
)
if "{search_term}" in endpoint:
search_term = url_quote(arguments.get("query", ""))
endpoint = endpoint.replace("{search_term}", search_term)
if self.use_browser:
return f"https://icd.who.int/browse/2024-01/mms{endpoint}"
return f"{ICD_API_BASE}{endpoint}"
_BOOL_PARAMS = ("flatResults", "useFlexisearch")
[docs]
def _make_request(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Make request to ICD API."""
access_token = self._get_access_token()
if not access_token:
return {
"status": "error",
"error": self._auth_error
or f"ICD API authentication required. {self._REGISTER}",
}
url = self._build_url(arguments)
headers = {
"Authorization": f"Bearer {access_token}",
"Accept": "application/json",
"Accept-Language": arguments.get("language", "en"),
"API-Version": "v2",
}
params = {}
for key in self._BOOL_PARAMS:
if key in arguments:
params[key] = str(arguments[key]).lower()
try:
resp = requests.get(url, headers=headers, params=params, timeout=30)
resp.raise_for_status()
return {
"status": "success",
"data": resp.json(),
"metadata": {
"source": "WHO ICD-11 API",
"endpoint": url,
"linearization": arguments.get("linearization", self.linearization),
"language": arguments.get("language", "en"),
},
}
except requests.exceptions.RequestException as e:
return {"status": "error", "error": f"Request failed: {e}"}
except ValueError as e:
return {"status": "error", "error": f"Failed to parse JSON: {e}"}
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute the tool with given arguments."""
return self._make_request(arguments)
[docs]
@register_tool("ICD10Tool")
class ICD10Tool(BaseTool):
"""Tool for ICD-10-CM codes using external API."""
[docs]
def __init__(self, tool_config):
super().__init__(tool_config)
fields = tool_config.get("fields", {})
self.endpoint_template = fields.get("endpoint", "")
self.base_url = fields.get("base_url", "https://clinicaltables.nlm.nih.gov/api")
[docs]
def _build_url(self, arguments: Dict[str, Any]) -> str:
"""Build the full URL from endpoint template and arguments."""
endpoint = self.endpoint_template
# Replace {code} placeholder
if "{code}" in endpoint:
code = arguments.get("code", "")
endpoint = endpoint.replace("{code}", code)
return f"{self.base_url}{endpoint}"
[docs]
@staticmethod
def _validate_pagination(arguments: Dict[str, Any]):
"""Validate Clinical Tables count/offset controls for ICD-10 search."""
limit = arguments.get("limit", 20)
offset = arguments.get("offset", 0)
if (
isinstance(limit, bool)
or not isinstance(limit, int)
or not 1 <= limit <= 500
):
return {
"status": "error",
"error": "limit must be an integer between 1 and 500",
"parameter": "limit",
}
if isinstance(offset, bool) or not isinstance(offset, int) or offset < 0:
return {
"status": "error",
"error": "offset must be a non-negative integer",
"parameter": "offset",
}
if offset + limit > 7500:
return {
"status": "error",
"error": (
"offset + limit must not exceed the Clinical Tables "
"retrieval limit of 7500"
),
"parameter": "offset",
}
return offset, limit
[docs]
def _make_request(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Make request to ICD-10 API (NLM Clinical Tables)."""
url = self._build_url(arguments)
# Build query parameters for search
params = {}
if "query" in arguments:
query = arguments.get("query")
if not isinstance(query, str) or not query.strip():
return {"status": "error", "error": "query is required"}
pagination_args = self._validate_pagination(arguments)
if isinstance(pagination_args, dict):
return pagination_args
offset, limit = pagination_args
params["sf"] = "code,name" # Search fields
params["df"] = "code,name"
params["terms"] = query.strip()
params["count"] = limit
params["offset"] = offset
try:
resp = requests.get(url, params=params, timeout=30)
resp.raise_for_status()
data = resp.json()
# Parse the response format from NLM Clinical Tables
# Format: [total_count, [codes], null, [[code, name], ...]]
if isinstance(data, list) and len(data) >= 4:
total = data[0]
results = data[3] if len(data) > 3 else []
formatted_results = []
for item in results:
if len(item) >= 2:
formatted_results.append({"code": item[0], "name": item[1]})
metadata = {
"source": "NLM Clinical Tables - ICD-10-CM",
"endpoint": url,
"version": "2026 ICD-10-CM codes",
"note": "ICD-10-CM is the US clinical modification of ICD-10",
}
if "query" in arguments:
try:
total = int(total)
except (TypeError, ValueError):
total = len(formatted_results)
accessible = min(total, 7500)
returned_count = len(formatted_results)
has_more = offset + returned_count < accessible
upstream_truncated = total > accessible
metadata.update(
{
"total_results": total,
"accessible_results": accessible,
"upstream_truncated": upstream_truncated,
"returned_count": returned_count,
"offset": offset,
"limit": limit,
"has_more": has_more,
"next_offset": (
offset + returned_count
if returned_count and has_more
else None
),
"truncated": has_more or upstream_truncated,
"result_set_complete": (
not has_more and not upstream_truncated
),
"warning": (
"NLM Clinical Tables limits retrieval to the first "
"7500 matches; refine the search to reach records "
"outside that window."
if upstream_truncated
else None
),
}
)
return {
"status": "success",
"data": {"total": total, "results": formatted_results},
"metadata": metadata,
}
# Direct code lookup
return {
"status": "success",
"data": data,
"metadata": {
"source": "NLM Clinical Tables - ICD-10-CM",
"endpoint": url,
},
}
except requests.exceptions.RequestException as e:
return {"status": "error", "error": f"Request failed: {str(e)}"}
except (ValueError, IndexError) as e:
return {"status": "error", "error": f"Failed to parse response: {str(e)}"}
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute the tool with given arguments."""
return self._make_request(arguments)