zhuchaowe/qwen_agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

pmda-drug-infomock数据

Browse Source
This commit is contained in:
朱潮 2026-05-11 19:15:03 +08:00
parent 5b634bc2ab
commit a92096a646
1 changed files with 313 additions and 376 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

689
skills/developing/pmda-drug-info/pmda_server.py
View File

@ -1,20 +1,15 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
PMDA drug information MCP server. PMDA drug information MCP server (mock data version).
Provides drug search, master info, interactions, restrictions, dosing, Provides drug search, master info, interactions, restrictions, dosing,
and full-text chapter retrieval via PostgreSQL + OpenSearch. and full-text chapter retrieval with mock data for testing.
""" """
import asyncio import asyncio
import json import json
import os
import sys import sys
from decimal import Decimal from typing import Any, Dict, Optional
from typing import Any, Dict, List, Optional
import psycopg2
import psycopg2.extras
from opensearchpy import OpenSearch
from mcp_common import ( from mcp_common import (
create_error_response, create_error_response,
@ -25,189 +20,269 @@ from mcp_common import (
handle_mcp_streaming, handle_mcp_streaming,
) )
# ---------------------------------------------------------------------------
# Configuration from environment variables
# ---------------------------------------------------------------------------
PG_DSN = os.getenv("PMDA_PG_DSN", "")
OS_HOST = os.getenv("PMDA_OS_HOST", "localhost")
OS_PORT = int(os.getenv("PMDA_OS_PORT", "9200"))
OS_INDEX = os.getenv("PMDA_OS_INDEX", "pmda_sections")
def _json_default(o):
"""JSON serializer for objects not serializable by default json code."""
if isinstance(o, Decimal):
return float(o)
raise TypeError(f"non-serializable: {type(o).__name__}")
def _dump(obj) -> str: def _dump(obj) -> str:
return json.dumps(obj, ensure_ascii=False, default=_json_default) return json.dumps(obj, ensure_ascii=False)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Lazy database connections # Mock data
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
_pg_conn = None
_os_client = None
# Drug lookup cache: yj_code -> (brand_name, yj_full) MOCK_DRUG_MASTER = {
_drug_lookup: Optional[Dict[str, tuple]] = None "2149039F1082": {
"yj_code": "2149039F1082",
"yj_full": "2149039F1082_1_17",
"brand_name": "ロサルタンK錠50mg「科研」",
"generic_name": "ロサルタンカリウム",
"category_code": "214",
"category_name": "アンジオテンシンII受容体拮抗薬",
"regulation": "劇薬, 処方箋医薬品",
"manufacturer": "科研製薬株式会社",
"revision_date": "2024-06",
},
"3399007H1021": {
"yj_code": "3399007H1021",
"yj_full": "3399007H1021_1_21",
"brand_name": "バイアスピリン錠100mg",
"generic_name": "アスピリン",
"category_code": "339",
"category_name": "血液・体液用薬",
"regulation": "処方箋医薬品",
"manufacturer": "バイエル薬品株式会社",
"revision_date": "2024-03",
},
"2179004F1026": {
"yj_code": "2179004F1026",
"yj_full": "2179004F1026_1_14",
"brand_name": "ノルバスク錠5mg",
"generic_name": "アムロジピンベシル酸塩",
"category_code": "217",
"category_name": "カルシウム拮抗薬",
"regulation": "処方箋医薬品",
"manufacturer": "ファイザー株式会社",
"revision_date": "2024-01",
},
}
MOCK_CATEGORIES = [
{"category_code": "214", "category_name": "アンジオテンシンII受容体拮抗薬", "level": "L2", "drug_count": 35},
{"category_code": "217", "category_name": "カルシウム拮抗薬", "level": "L2", "drug_count": 48},
{"category_code": "339", "category_name": "血液・体液用薬", "level": "L2", "drug_count": 22},
{"category_code": "612", "category_name": "消化性潰瘍用剤", "level": "L2", "drug_count": 40},
]
def _get_pg(): MOCK_INTERACTIONS = [
global _pg_conn {
if _pg_conn is None or _pg_conn.closed: "drug_a_yj": "2149039F1082",
if not PG_DSN: "drug_b_yj": "3399007H1021",
raise RuntimeError("PMDA_PG_DSN environment variable is not set") "drug_b_class": "アスピリン（抗血小板剤）",
_pg_conn = psycopg2.connect(PG_DSN) "severity": "併用注意",
_pg_conn.autocommit = True "mechanism": "ARBの降圧作用を減弱するおそれがある。また、腎機能低下・高カリウム血症のリスクを増大。",
return _pg_conn "clinical_effect": "降圧効果の減弱、腎機能悪化、高カリウム血症に注意。",
"source_drug_yj": "2149039F1082",
"source_section": "10.2 併用注意",
},
{
"drug_a_yj": "3399007H1021",
"drug_b_yj": "2149039F1082",
"drug_b_class": "ロサルタンカリウム（ARB）",
"severity": "併用注意",
"mechanism": "アスピリンの副作用（消化性潰瘍、腎機能低下）を増強するおそれ。",
"clinical_effect": "消化性潰瘍、腎機能低下に注意。血清カリウム値の上昇に注意。",
"source_drug_yj": "3399007H1021",
"source_section": "10.2 併用注意",
},
]
MOCK_RESTRICTIONS = [
{
"drug_yj": "2149039F1082",
"condition_type": "腎機能障害",
"condition_text": "腎機能障害患者",
"condition_params": {"eGFR_max": 30},
"severity": "慎重投与",
"source_section": "9.2 腎機能障害患者",
},
{
"drug_yj": "2149039F1082",
"condition_type": "妊婦",
"condition_text": "妊娠中の女性",
"condition_params": {},
"severity": "禁忌",
"source_section": "9.5 妊婦",
},
{
"drug_yj": "2149039F1082",
"condition_type": "高齢者",
"condition_text": "高齢者（65歳以上）",
"condition_params": {},
"severity": "慎重投与",
"source_section": "9.8 高齢者",
},
{
"drug_yj": "3399007H1021",
"condition_type": "過敏症",
"condition_text": "本剤の成分に対し過敏症の既往歴のある患者",
"condition_params": {},
"severity": "禁忌",
"source_section": "2. 禁忌",
},
]
def _get_os() -> OpenSearch: MOCK_DOSING = [
global _os_client {
if _os_client is None: "drug_yj": "2149039F1082",
_os_client = OpenSearch( "patient_segment": "成人",
hosts=[{"host": OS_HOST, "port": OS_PORT}], "segment_params": {},
use_ssl=False, "indication_code": "高血圧症",
verify_certs=False, "dose_amount": "50",
) "dose_unit": "mg",
return _os_client "frequency": "1日1回",
"duration": "",
"adjustment_text": "効果不十分な場合は100mgまで増量可",
"source_section": "6. 用法及び用量",
},
{
"drug_yj": "2149039F1082",
"patient_segment": "腎機能障害患者",
"segment_params": {"eGFR_max": 30},
"indication_code": "高血圧症",
"dose_amount": "25",
"dose_unit": "mg",
"frequency": "1日1回",
"duration": "",
"adjustment_text": "eGFR 30以下では用量を減ずること。血清カリウム・クレアチニンの推移に注意。",
"source_section": "9.2 腎機能障害患者",
},
]
MOCK_CHAPTERS = {
"2149039F1082_1_17": [
{"section_title": "1. 警告", "line_num": 1, "text_len": 120},
{"section_title": "2. 禁忌", "line_num": 5, "text_len": 80},
{"section_title": "4. 効能・効果", "line_num": 12, "text_len": 60},
{"section_title": "6. 用法及び用量", "line_num": 20, "text_len": 150},
{"section_title": "9.2 腎機能障害患者", "line_num": 45, "text_len": 200},
{"section_title": "9.5 妊婦", "line_num": 52, "text_len": 180},
{"section_title": "9.8 高齢者", "line_num": 60, "text_len": 100},
{"section_title": "10.2 併用注意", "line_num": 75, "text_len": 350},
{"section_title": "11.1 重大な副作用", "line_num": 90, "text_len": 400},
{"section_title": "11.2 その他の副作用", "line_num": 110, "text_len": 300},
],
"3399007H1021_1_21": [
{"section_title": "1. 警告", "line_num": 1, "text_len": 100},
{"section_title": "2. 禁忌", "line_num": 4, "text_len": 90},
{"section_title": "4. 効能・効果", "line_num": 10, "text_len": 55},
{"section_title": "6. 用法及び用量", "line_num": 18, "text_len": 130},
{"section_title": "10.2 併用注意", "line_num": 70, "text_len": 300},
{"section_title": "11.1 重大な副作用", "line_num": 85, "text_len": 450},
{"section_title": "11.2 その他の副作用", "line_num": 105, "text_len": 280},
],
}
def _load_drug_lookup() -> Dict[str, tuple]: MOCK_SECTION_TEXT = {
"""Load yj_code -> (brand_name, yj_full) mapping from drug_master.""" ("2149039F1082_1_17", "9.2 腎機能障害患者"): (
global _drug_lookup "9.2 腎機能障害患者\n"
if _drug_lookup is not None: "腎機能障害患者（eGFR 30 mL/min/1.73m²以下）には、ロサルタンカリウムの"
return _drug_lookup "投与開始用量を25mg/日とし、血清カリウム及び血清クレアチニンの推移に"
conn = _get_pg() "十分注意すること。\n"
with conn.cursor() as cur: "【理由】腎機能障害患者では、本剤の投与により急速に腎機能が悪化する"
cur.execute("SELECT yj_code, brand_name, yj_full FROM drug_master") "おそれがある。また、高カリウム血症があらわれやすい。"
_drug_lookup = { ),
row[0]: (row[1] or "", row[2] or row[0]) for row in cur.fetchall() ("2149039F1082_1_17", "9.5 妊婦"): (
} "9.5 妊婦\n"
return _drug_lookup "妊婦又は妊娠している可能性のある女性には投与しないこと。\n"
"【理由】妊娠中期・末期にレニン-アンジオテンシン系に作用する薬剤を"
"投与された患者では、胎児の腎機能低下、羊水過少症、頭蓋の発育不全、"
"肺低形成等があらわれるおそれがある。"
),
("2149039F1082_1_17", "10.2 併用注意"): (
"10.2 併用注意\n"
"・アスピリン（抗血小板剤）\n"
" 【リスク】ARBの降圧作用を減弱するおそれがある。\n"
" 腎機能低下・高カリウム血症のリスクを増大。\n"
" 【措置】降圧効果の減弱、腎機能悪化、高カリウム血症に注意すること。"
),
("2149039F1082_1_17", "11.1 重大な副作用"): (
"11.1 重大な副作用\n"
"・血管浮腫（頻度不明）：顔面、口唇、咽頭、舌等の腫脹があらわれた場合には"
"直ちに投与を中止し、適切な処置を行うこと。\n"
"・高カリウム血症（0.1%未満）：血清カリウム値の上昇があらわれることがある。\n"
"・腎機能悪化（0.1%未満）：BUN、クレアチニンの上昇があらわれることがある。"
),
("3399007H1021_1_21", "10.2 併用注意"): (
"10.2 併用注意\n"
"・ロサルタンカリウム（ARB）\n"
" 【リスク】アスピリンの副作用（消化性潰瘍、腎機能低下）を増強するおそれ。\n"
" 【措置】消化性潰瘍、腎機能低下に注意。血清カリウム値の上昇に注意すること。"
),
("3399007H1021_1_21", "11.1 重大な副作用"): (
"11.1 重大な副作用\n"
"・ショック、アナフィラキシー（頻度不明）：呼吸困難、血圧低下等があらわれた\n"
" 場合には直ちに投与を中止し、適切な処置を行うこと。\n"
"・消化性潰瘍（0.1%未満）：出血、穿孔があらわれることがある。\n"
"・腎機能障害（0.1%未満）：急性腎不全があらわれることがある。"
),
}
def _citation(drug_yj: str, section: Optional[str]) -> str: def _citation(drug_yj: str, section: Optional[str]) -> str:
"""Format citation string: [出典: <brand> (yj_full=<id>) / <section>]""" drug = MOCK_DRUG_MASTER.get(drug_yj, {})
lk = _load_drug_lookup() brand = drug.get("brand_name", "")
brand, yj_full = lk.get(drug_yj, ("", drug_yj)) yj_full = drug.get("yj_full", drug_yj)
chap = section or "（章不明）" chap = section or "（章不明）"
return f"[出典: {brand} (yj_full={yj_full}) / {chap}]" return f"[出典: {brand} (yj_full={yj_full}) / {chap}]"
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Tool implementations # Tool implementations (mock)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def _tool_search_drugs(query: str, kind: str = "auto", limit: int = 10) -> str: def _tool_search_drugs(query: str, kind: str = "auto", limit: int = 10) -> str:
"""Search drugs by brand name, generic name, or YJ code.""" results = []
conn = _get_pg() for code, d in MOCK_DRUG_MASTER.items():
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: q = query.lower()
if kind == "yj": if (kind == "brand" and q in d["brand_name"].lower()) or \
cur.execute( (kind == "generic" and q in d["generic_name"].lower()) or \
"SELECT yj_full, yj_code, brand_name, generic_name, " (kind == "yj" and (q in d["yj_code"].lower() or q in d["yj_full"].lower())) or \
"category_code, category_name FROM drug_master " (kind == "auto" and (q in d["brand_name"].lower() or q in d["generic_name"].lower()
"WHERE yj_code ILIKE %s OR yj_full ILIKE %s LIMIT %s", or q in d["yj_code"].lower() or q in d["yj_full"].lower())):
(f"%{query}%", f"%{query}%", limit), results.append({
) "yj_full": d["yj_full"],
elif kind == "brand": "yj_code": d["yj_code"],
cur.execute( "brand": d["brand_name"],
"SELECT yj_full, yj_code, brand_name, generic_name, " "generic": d["generic_name"],
"category_code, category_name, " "category": f"{d['category_code']} {d['category_name']}",
"similarity(brand_name, %s) AS score " "score": 1.0,
"FROM drug_master " })
"WHERE brand_name ILIKE %s ORDER BY score DESC LIMIT %s", return _dump(results[:limit])
(query, f"%{query}%", limit),
)
elif kind == "generic":
cur.execute(
"SELECT yj_full, yj_code, brand_name, generic_name, "
"category_code, category_name, "
"similarity(generic_name, %s) AS score "
"FROM drug_master "
"WHERE generic_name ILIKE %s ORDER BY score DESC LIMIT %s",
(query, f"%{query}%", limit),
)
else: # auto
cur.execute(
"SELECT yj_full, yj_code, brand_name, generic_name, "
"category_code, category_name, "
"GREATEST("
" similarity(brand_name, %s),"
" similarity(generic_name, %s)"
") AS score "
"FROM drug_master "
"WHERE brand_name ILIKE %s OR generic_name ILIKE %s "
" OR yj_code ILIKE %s OR yj_full ILIKE %s "
"ORDER BY score DESC LIMIT %s",
(query, query, f"%{query}%", f"%{query}%", f"%{query}%", f"%{query}%", limit),
)
rows = cur.fetchall()
return _dump([
{
"yj_full": r.get("yj_full", ""),
"yj_code": r.get("yj_code", ""),
"brand": r.get("brand_name", ""),
"generic": r.get("generic_name", ""),
"category": f"{r.get('category_code', '')} {r.get('category_name', '')}".strip(),
"score": float(r.get("score", 0)) if r.get("score") else 0.0,
}
for r in rows
])
def _tool_list_categories() -> str: def _tool_list_categories() -> str:
"""List all L1/L2 drug categories with drug counts.""" return _dump(MOCK_CATEGORIES)
conn = _get_pg()
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
cur.execute(
"SELECT c.category_code, c.category_name, c.level, "
"COUNT(m.yj_code) AS drug_count "
"FROM drug_category c "
"LEFT JOIN drug_master m ON m.category_code = c.category_code "
"WHERE c.level IN ('L1', 'L2') "
"GROUP BY c.category_code, c.category_name, c.level "
"ORDER BY c.category_code"
)
rows = cur.fetchall()
return _dump([dict(r) for r in rows])
def _tool_list_drugs_in_category(l2_code: str, limit_generics: int = 50) -> str: def _tool_list_drugs_in_category(l2_code: str, limit_generics: int = 50) -> str:
"""List drugs under a specific L2 category code.""" results = []
conn = _get_pg() seen_generics = set()
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: for code, d in MOCK_DRUG_MASTER.items():
cur.execute( if d["category_code"].startswith(l2_code) and d["generic_name"] not in seen_generics:
"SELECT generic_name, json_agg(json_build_object(" seen_generics.add(d["generic_name"])
" 'yj_code', yj_code, 'brand_name', brand_name, 'yj_full', yj_full" results.append({
")) AS brands " "generic_name": d["generic_name"],
"FROM drug_master " "brands": [{"yj_code": d["yj_code"], "brand_name": d["brand_name"], "yj_full": d["yj_full"]}],
"WHERE category_code ILIKE %s " })
"GROUP BY generic_name " return _dump(results[:limit_generics])
"ORDER BY generic_name LIMIT %s",
(f"{l2_code}%", limit_generics),
)
rows = cur.fetchall()
return _dump([{"generic_name": r["generic_name"], "brands": r["brands"]} for r in rows])
def _tool_get_drug_master(yj_code: str) -> str: def _tool_get_drug_master(yj_code: str) -> str:
"""Get basic info for a drug by yj_code.""" d = MOCK_DRUG_MASTER.get(yj_code)
conn = _get_pg() if not d:
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
cur.execute(
"SELECT * FROM drug_master WHERE yj_code = %s LIMIT 1",
(yj_code,),
)
row = cur.fetchone()
if not row:
return _dump({"error": f"yj_code {yj_code} not found"}) return _dump({"error": f"yj_code {yj_code} not found"})
d = dict(row) result = dict(d)
d["_citation"] = f"[出典: {row.get('brand_name', '')} (yj_full={row.get('yj_full', '')}) / 添付文書冒頭]" result["_citation"] = f"[出典: {d['brand_name']} (yj_full={d['yj_full']}) / 添付文書冒頭]"
return _dump(d) return _dump(result)
def _tool_get_drug_interactions( def _tool_get_drug_interactions(
@ -217,45 +292,22 @@ def _tool_get_drug_interactions(
keyword: Optional[str] = None, keyword: Optional[str] = None,
limit: int = 30, limit: int = 30,
) -> str: ) -> str:
"""Search drug_interaction table.""" results = []
conditions = [] for r in MOCK_INTERACTIONS:
params = [] if drug_a_yj and r["drug_a_yj"] != drug_a_yj:
if drug_a_yj: continue
conditions.append("drug_a_yj = %s") if drug_b_yj and r["drug_b_yj"] != drug_b_yj:
params.append(drug_a_yj) continue
if drug_b_yj: if severity and r["severity"] != severity:
conditions.append("(drug_b_yj = %s OR drug_a_yj = %s)") continue
params.extend([drug_b_yj, drug_b_yj]) if keyword and keyword.lower() not in (
if severity: (r.get("drug_b_class") or "").lower()
conditions.append("severity = %s") + (r.get("mechanism") or "").lower()
params.append(severity) + (r.get("clinical_effect") or "").lower()
if keyword: ):
conditions.append("(drug_b_class ILIKE %s OR mechanism ILIKE %s OR clinical_effect ILIKE %s)") continue
k = f"%{keyword}%" results.append({**r, "_citation": _citation(r["source_drug_yj"], r["source_section"])})
params.extend([k, k, k]) return _dump(results[:limit])
where = " AND ".join(conditions) if conditions else "1=1"
conn = _get_pg()
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
cur.execute(
f"SELECT * FROM drug_interaction WHERE {where} LIMIT %s",
(*params, limit),
)
rows = cur.fetchall()
return _dump([
{
"drug_a_yj": r.get("drug_a_yj"),
"drug_b_yj": r.get("drug_b_yj"),
"drug_b_class": r.get("drug_b_class"),
"severity": r.get("severity"),
"mechanism": r.get("mechanism"),
"clinical_effect": r.get("clinical_effect"),
"source_drug_yj": r.get("source_drug_yj"),
"source_section": r.get("source_section"),
"_citation": _citation(r.get("source_drug_yj", ""), r.get("source_section")),
}
for r in rows
])
def _tool_get_drug_restrictions( def _tool_get_drug_restrictions(
@ -265,42 +317,18 @@ def _tool_get_drug_restrictions(
keyword: Optional[str] = None, keyword: Optional[str] = None,
limit: int = 30, limit: int = 30,
) -> str: ) -> str:
"""Search drug_restriction table.""" results = []
conditions = [] for r in MOCK_RESTRICTIONS:
params = [] if drug_yj and r["drug_yj"] != drug_yj:
if drug_yj: continue
conditions.append("drug_yj = %s") if condition_type and r["condition_type"] != condition_type:
params.append(drug_yj) continue
if condition_type: if severity and r["severity"] != severity:
conditions.append("condition_type = %s") continue
params.append(condition_type) if keyword and keyword.lower() not in (r.get("condition_text") or "").lower():
if severity: continue
conditions.append("severity = %s") results.append({**r, "_citation": _citation(r["drug_yj"], r["source_section"])})
params.append(severity) return _dump(results[:limit])
if keyword:
conditions.append("condition_text ILIKE %s")
params.append(f"%{keyword}%")
where = " AND ".join(conditions) if conditions else "1=1"
conn = _get_pg()
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
cur.execute(
f"SELECT * FROM drug_restriction WHERE {where} LIMIT %s",
(*params, limit),
)
rows = cur.fetchall()
return _dump([
{
"drug_yj": r.get("drug_yj"),
"condition_type": r.get("condition_type"),
"condition_text": r.get("condition_text"),
"condition_params": r.get("condition_params"),
"severity": r.get("severity"),
"source_section": r.get("source_section"),
"_citation": _citation(r.get("drug_yj", ""), r.get("source_section")),
}
for r in rows
])
def _tool_get_drug_dosing( def _tool_get_drug_dosing(
@ -308,36 +336,14 @@ def _tool_get_drug_dosing(
patient_segment: Optional[str] = None, patient_segment: Optional[str] = None,
limit: int = 20, limit: int = 20,
) -> str: ) -> str:
"""Search drug_dosing table.""" results = []
conditions = ["drug_yj = %s"] for r in MOCK_DOSING:
params = [drug_yj] if r["drug_yj"] != drug_yj:
if patient_segment: continue
conditions.append("patient_segment = %s") if patient_segment and r["patient_segment"] != patient_segment:
params.append(patient_segment) continue
results.append({**r, "_citation": _citation(drug_yj, r["source_section"])})
where = " AND ".join(conditions) return _dump(results[:limit])
conn = _get_pg()
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
cur.execute(
f"SELECT * FROM drug_dosing WHERE {where} LIMIT %s",
(*params, limit),
)
rows = cur.fetchall()
return _dump([
{
"patient_segment": r.get("patient_segment"),
"segment_params": r.get("segment_params"),
"indication_code": r.get("indication_code"),
"dose_amount": r.get("dose_amount"),
"dose_unit": r.get("dose_unit"),
"frequency": r.get("frequency"),
"duration": r.get("duration"),
"adjustment_text": r.get("adjustment_text"),
"source_section": r.get("source_section"),
"_citation": _citation(drug_yj, r.get("source_section")),
}
for r in rows
])
def _tool_search_section_text( def _tool_search_section_text(
@ -345,141 +351,73 @@ def _tool_search_section_text(
section_filter: str = "", section_filter: str = "",
limit: int = 30, limit: int = 30,
) -> str: ) -> str:
"""Full-text search in OpenSearch pmda_sections index."""
if not keyword.strip(): if not keyword.strip():
return _dump({"keyword": keyword, "total_drugs": 0, "shown": 0, "hits": []}) return _dump({"keyword": keyword, "total_drugs": 0, "shown": 0, "hits": []})
size = min(max(1, limit), 100) # Simple mock: search through section text
body: Dict[str, Any] = {
"size": size,
"_source": ["yj_full", "brand_names", "generic_name", "l2_code", "l2_name", "section_title", "line_num"],
"query": {"bool": {"must": [{"match": {"text": keyword}}]}},
"collapse": {
"field": "yj_full",
"inner_hits": {
"name": "matches",
"size": 2,
"_source": ["section_title", "line_num"],
"highlight": {"fields": {"text": {"fragment_size": 160, "number_of_fragments": 1}}},
},
},
"aggs": {"total_drugs": {"cardinality": {"field": "yj_full"}}},
}
if section_filter:
body["query"]["bool"]["filter"] = [
{"wildcard": {"section_title.raw": f"*{section_filter}*"}}
]
client = _get_os()
resp = client.search(index=OS_INDEX, body=body)
total = int(resp["aggregations"]["total_drugs"]["value"])
hits_out = [] hits_out = []
for h in resp["hits"]["hits"]: for (yj_full, section_title), text in MOCK_SECTION_TEXT.items():
src = h.get("_source") or {} if section_filter and section_filter not in section_title:
inner = h.get("inner_hits", {}).get("matches", {}).get("hits", {}).get("hits", []) continue
matches = [] if keyword.lower() in text.lower():
seen = set() drug = None
for ih in inner: for d in MOCK_DRUG_MASTER.values():
ih_src = ih.get("_source") or {} if d["yj_full"] == yj_full:
title = ih_src.get("section_title") or "" drug = d
if title in seen: break
if not drug:
continue continue
seen.add(title) brand = drug["brand_name"]
hl = ih.get("highlight", {}).get("text", [""]) # Deduplicate by yj_full
matches.append({"section_title": title, "snippet": hl[0] if hl else ""}) existing = [h for h in hits_out if h["yj_full"] == yj_full]
brand = (src.get("brand_names") or [""])[0] if existing:
yj_full = src.get("yj_full") or "" existing[0]["matches"].append({
hits_out.append({ "section_title": section_title,
"yj_full": yj_full, "snippet": text[:160],
"brand": brand, })
"generic": src.get("generic_name") or "", continue
"l2": f"{src.get('l2_code') or ''} {src.get('l2_name') or ''}".strip(), hits_out.append({
"matches": matches, "yj_full": yj_full,
"_citation_template": f"[出典: {brand} (yj_full={yj_full}) / <該当章>]", "brand": brand,
}) "generic": drug["generic_name"],
"l2": f"{drug['category_code']} {drug['category_name']}",
"matches": [{"section_title": section_title, "snippet": text[:160]}],
"_citation_template": f"[出典: {brand} (yj_full={yj_full}) / <該当章>]",
})
out = { return _dump({
"keyword": keyword, "keyword": keyword,
"section_filter": section_filter or None, "section_filter": section_filter or None,
"total_drugs": total, "total_drugs": len({h["yj_full"] for h in hits_out}),
"shown": len(hits_out), "shown": len(hits_out),
"hits": hits_out, "hits": hits_out[:limit],
} })
if total > len(hits_out):
out["_more_count"] = total - len(hits_out)
return _dump(out)
def _tool_list_drug_chapters(yj_full: str) -> str: def _tool_list_drug_chapters(yj_full: str) -> str:
"""List all chapter titles for a drug's package insert.""" sections = MOCK_CHAPTERS.get(yj_full)
client = _get_os() if not sections:
body = {
"size": 200,
"_source": ["yj_full", "brand_names", "generic_name", "section_title", "line_num"],
"query": {"term": {"yj_full": yj_full}},
"sort": [{"line_num": {"order": "asc"}}],
}
resp = client.search(index=OS_INDEX, body=body)
hits = resp["hits"]["hits"]
if not hits:
return _dump({"error": f"yj_full {yj_full} の章節が見つかりません。"}) return _dump({"error": f"yj_full {yj_full} の章節が見つかりません。"})
sections = [] drug = None
for h in hits: for d in MOCK_DRUG_MASTER.values():
src = h.get("_source") or {} if d["yj_full"] == yj_full:
# Calculate text length from _score or use stored field drug = d
sections.append({ break
"section_title": src.get("section_title", ""),
"line_num": src.get("line_num", 0),
"text_len": 0, # not available from list query
})
head = hits[0].get("_source") or {}
return _dump({ return _dump({
"yj_full": yj_full, "yj_full": yj_full,
"brand": (head.get("brand_names") or [""])[0], "brand": drug["brand_name"] if drug else "",
"generic": head.get("generic_name", ""), "generic": drug["generic_name"] if drug else "",
"n_sections": len(sections), "n_sections": len(sections),
"sections": sections, "sections": sections,
}) })
def _tool_read_drug_chapter(yj_full: str, section_title: str) -> str: def _tool_read_drug_chapter(yj_full: str, section_title: str) -> str:
"""Read verbatim text of a specific chapter.""" text = MOCK_SECTION_TEXT.get((yj_full, section_title))
client = _get_os() if text:
body = { return text[:8000]
"size": 1,
"_source": ["text", "section_title"],
"query": {
"bool": {
"must": [
{"term": {"yj_full": yj_full}},
{"term": {"section_title.keyword": section_title}},
]
}
},
}
resp = client.search(index=OS_INDEX, body=body)
hits = resp["hits"]["hits"]
if hits:
text = hits[0].get("_source", {}).get("text", "")
if text:
return text[:8000]
# Fallback: try match instead of term for section_title
body["query"]["bool"]["must"][1] = {"match_phrase": {"section_title": section_title}}
resp = client.search(index=OS_INDEX, body=body)
hits = resp["hits"]["hits"]
if hits:
text = hits[0].get("_source", {}).get("text", "")
if text:
return text[:8000]
# Not found — suggest listing chapters
return _dump({ return _dump({
"error": f"section_title {section_title!r} は {yj_full} に存在しません。", "error": f"section_title {section_title!r} は {yj_full} に存在しません。",
"hint": "list_drug_chapters で取得した sections[].section_title をそのまま渡してください。", "hint": "list_drug_chapters で取得した sections[].section_title をそのまま渡してください。",
@ -490,7 +428,6 @@ def _tool_read_drug_chapter(yj_full: str, section_title: str) -> str:
# MCP request handler # MCP request handler
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Map tool names to their implementation functions
_TOOL_DISPATCH = { _TOOL_DISPATCH = {
"search_drugs": lambda args: _tool_search_drugs( "search_drugs": lambda args: _tool_search_drugs(
query=args.get("query", ""), query=args.get("query", ""),