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

Merge branch 'feature/moshui20260511-sub-agents' into dev

Browse Source
This commit is contained in:
朱潮 2026-05-11 20:16:29 +08:00
commit 9eb1d6b068
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
"""
PMDA drug information MCP server.
PMDA drug information MCP server (mock data version).
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 json
import os
import sys
from decimal import Decimal
from typing import Any, Dict, List, Optional
import psycopg2
import psycopg2.extras
from opensearchpy import OpenSearch
from typing import Any, Dict, Optional
from mcp_common import (
create_error_response,
@ -25,189 +20,269 @@ from mcp_common import (
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:
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)
_drug_lookup: Optional[Dict[str, tuple]] = None
MOCK_DRUG_MASTER = {
"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():
global _pg_conn
if _pg_conn is None or _pg_conn.closed:
if not PG_DSN:
raise RuntimeError("PMDA_PG_DSN environment variable is not set")
_pg_conn = psycopg2.connect(PG_DSN)
_pg_conn.autocommit = True
return _pg_conn
MOCK_INTERACTIONS = [
{
"drug_a_yj": "2149039F1082",
"drug_b_yj": "3399007H1021",
"drug_b_class": "アスピリン（抗血小板剤）",
"severity": "併用注意",
"mechanism": "ARBの降圧作用を減弱するおそれがある。また、腎機能低下・高カリウム血症のリスクを増大。",
"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:
global _os_client
if _os_client is None:
_os_client = OpenSearch(
hosts=[{"host": OS_HOST, "port": OS_PORT}],
use_ssl=False,
verify_certs=False,
)
return _os_client
MOCK_DOSING = [
{
"drug_yj": "2149039F1082",
"patient_segment": "成人",
"segment_params": {},
"indication_code": "高血圧症",
"dose_amount": "50",
"dose_unit": "mg",
"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]:
"""Load yj_code -> (brand_name, yj_full) mapping from drug_master."""
global _drug_lookup
if _drug_lookup is not None:
return _drug_lookup
conn = _get_pg()
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()
}
return _drug_lookup
MOCK_SECTION_TEXT = {
("2149039F1082_1_17", "9.2 腎機能障害患者"): (
"9.2 腎機能障害患者\n"
"腎機能障害患者（eGFR 30 mL/min/1.73m²以下）には、ロサルタンカリウムの"
"投与開始用量を25mg/日とし、血清カリウム及び血清クレアチニンの推移に"
"十分注意すること。\n"
"【理由】腎機能障害患者では、本剤の投与により急速に腎機能が悪化する"
"おそれがある。また、高カリウム血症があらわれやすい。"
),
("2149039F1082_1_17", "9.5 妊婦"): (
"9.5 妊婦\n"
"妊婦又は妊娠している可能性のある女性には投与しないこと。\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:
"""Format citation string: [出典: <brand> (yj_full=<id>) / <section>]"""
lk = _load_drug_lookup()
brand, yj_full = lk.get(drug_yj, ("", drug_yj))
drug = MOCK_DRUG_MASTER.get(drug_yj, {})
brand = drug.get("brand_name", "")
yj_full = drug.get("yj_full", drug_yj)
chap = section or "（章不明）"
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:
"""Search drugs by brand name, generic name, or YJ code."""
conn = _get_pg()
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
if kind == "yj":
cur.execute(
"SELECT yj_full, yj_code, brand_name, generic_name, "
"category_code, category_name FROM drug_master "
"WHERE yj_code ILIKE %s OR yj_full ILIKE %s LIMIT %s",
(f"%{query}%", f"%{query}%", limit),
)
elif kind == "brand":
cur.execute(
"SELECT yj_full, yj_code, brand_name, generic_name, "
"category_code, category_name, "
"similarity(brand_name, %s) AS score "
"FROM drug_master "
"WHERE brand_name ILIKE %s ORDER BY score DESC LIMIT %s",
(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
])
results = []
for code, d in MOCK_DRUG_MASTER.items():
q = query.lower()
if (kind == "brand" and q in d["brand_name"].lower()) or \
(kind == "generic" and q in d["generic_name"].lower()) or \
(kind == "yj" and (q in d["yj_code"].lower() or q in d["yj_full"].lower())) or \
(kind == "auto" and (q in d["brand_name"].lower() or q in d["generic_name"].lower()
or q in d["yj_code"].lower() or q in d["yj_full"].lower())):
results.append({
"yj_full": d["yj_full"],
"yj_code": d["yj_code"],
"brand": d["brand_name"],
"generic": d["generic_name"],
"category": f"{d['category_code']} {d['category_name']}",
"score": 1.0,
})
return _dump(results[:limit])
def _tool_list_categories() -> str:
"""List all L1/L2 drug categories with drug counts."""
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])
return _dump(MOCK_CATEGORIES)
def _tool_list_drugs_in_category(l2_code: str, limit_generics: int = 50) -> str:
"""List drugs under a specific L2 category code."""
conn = _get_pg()
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
cur.execute(
"SELECT generic_name, json_agg(json_build_object("
" 'yj_code', yj_code, 'brand_name', brand_name, 'yj_full', yj_full"
")) AS brands "
"FROM drug_master "
"WHERE category_code ILIKE %s "
"GROUP BY generic_name "
"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])
results = []
seen_generics = set()
for code, d in MOCK_DRUG_MASTER.items():
if d["category_code"].startswith(l2_code) and d["generic_name"] not in seen_generics:
seen_generics.add(d["generic_name"])
results.append({
"generic_name": d["generic_name"],
"brands": [{"yj_code": d["yj_code"], "brand_name": d["brand_name"], "yj_full": d["yj_full"]}],
})
return _dump(results[:limit_generics])
def _tool_get_drug_master(yj_code: str) -> str:
"""Get basic info for a drug by yj_code."""
conn = _get_pg()
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:
d = MOCK_DRUG_MASTER.get(yj_code)
if not d:
return _dump({"error": f"yj_code {yj_code} not found"})
d = dict(row)
d["_citation"] = f"[出典: {row.get('brand_name', '')} (yj_full={row.get('yj_full', '')}) / 添付文書冒頭]"
return _dump(d)
result = dict(d)
result["_citation"] = f"[出典: {d['brand_name']} (yj_full={d['yj_full']}) / 添付文書冒頭]"
return _dump(result)
def _tool_get_drug_interactions(
@ -217,45 +292,22 @@ def _tool_get_drug_interactions(
keyword: Optional[str] = None,
limit: int = 30,
) -> str:
"""Search drug_interaction table."""
conditions = []
params = []
if drug_a_yj:
conditions.append("drug_a_yj = %s")
params.append(drug_a_yj)
if drug_b_yj:
conditions.append("(drug_b_yj = %s OR drug_a_yj = %s)")
params.extend([drug_b_yj, drug_b_yj])
if severity:
conditions.append("severity = %s")
params.append(severity)
if keyword:
conditions.append("(drug_b_class ILIKE %s OR mechanism ILIKE %s OR clinical_effect ILIKE %s)")
k = f"%{keyword}%"
params.extend([k, k, k])
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
])
results = []
for r in MOCK_INTERACTIONS:
if drug_a_yj and r["drug_a_yj"] != drug_a_yj:
continue
if drug_b_yj and r["drug_b_yj"] != drug_b_yj:
continue
if severity and r["severity"] != severity:
continue
if keyword and keyword.lower() not in (
(r.get("drug_b_class") or "").lower()
+ (r.get("mechanism") or "").lower()
+ (r.get("clinical_effect") or "").lower()
):
continue
results.append({**r, "_citation": _citation(r["source_drug_yj"], r["source_section"])})
return _dump(results[:limit])
def _tool_get_drug_restrictions(
@ -265,42 +317,18 @@ def _tool_get_drug_restrictions(
keyword: Optional[str] = None,
limit: int = 30,
) -> str:
"""Search drug_restriction table."""
conditions = []
params = []
if drug_yj:
conditions.append("drug_yj = %s")
params.append(drug_yj)
if condition_type:
conditions.append("condition_type = %s")
params.append(condition_type)
if severity:
conditions.append("severity = %s")
params.append(severity)
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
])
results = []
for r in MOCK_RESTRICTIONS:
if drug_yj and r["drug_yj"] != drug_yj:
continue
if condition_type and r["condition_type"] != condition_type:
continue
if severity and r["severity"] != severity:
continue
if keyword and keyword.lower() not in (r.get("condition_text") or "").lower():
continue
results.append({**r, "_citation": _citation(r["drug_yj"], r["source_section"])})
return _dump(results[:limit])
def _tool_get_drug_dosing(
@ -308,36 +336,14 @@ def _tool_get_drug_dosing(
patient_segment: Optional[str] = None,
limit: int = 20,
) -> str:
"""Search drug_dosing table."""
conditions = ["drug_yj = %s"]
params = [drug_yj]
if patient_segment:
conditions.append("patient_segment = %s")
params.append(patient_segment)
where = " AND ".join(conditions)
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
])
results = []
for r in MOCK_DOSING:
if r["drug_yj"] != drug_yj:
continue
if patient_segment and r["patient_segment"] != patient_segment:
continue
results.append({**r, "_citation": _citation(drug_yj, r["source_section"])})
return _dump(results[:limit])
def _tool_search_section_text(
@ -345,141 +351,73 @@ def _tool_search_section_text(
section_filter: str = "",
limit: int = 30,
) -> str:
"""Full-text search in OpenSearch pmda_sections index."""
if not keyword.strip():
return _dump({"keyword": keyword, "total_drugs": 0, "shown": 0, "hits": []})
size = min(max(1, limit), 100)
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"])
# Simple mock: search through section text
hits_out = []
for h in resp["hits"]["hits"]:
src = h.get("_source") or {}
inner = h.get("inner_hits", {}).get("matches", {}).get("hits", {}).get("hits", [])
matches = []
seen = set()
for ih in inner:
ih_src = ih.get("_source") or {}
title = ih_src.get("section_title") or ""
if title in seen:
for (yj_full, section_title), text in MOCK_SECTION_TEXT.items():
if section_filter and section_filter not in section_title:
continue
if keyword.lower() in text.lower():
drug = None
for d in MOCK_DRUG_MASTER.values():
if d["yj_full"] == yj_full:
drug = d
break
if not drug:
continue
seen.add(title)
hl = ih.get("highlight", {}).get("text", [""])
matches.append({"section_title": title, "snippet": hl[0] if hl else ""})
brand = (src.get("brand_names") or [""])[0]
yj_full = src.get("yj_full") or ""
hits_out.append({
"yj_full": yj_full,
"brand": brand,
"generic": src.get("generic_name") or "",
"l2": f"{src.get('l2_code') or ''} {src.get('l2_name') or ''}".strip(),
"matches": matches,
"_citation_template": f"[出典: {brand} (yj_full={yj_full}) / <該当章>]",
})
brand = drug["brand_name"]
# Deduplicate by yj_full
existing = [h for h in hits_out if h["yj_full"] == yj_full]
if existing:
existing[0]["matches"].append({
"section_title": section_title,
"snippet": text[:160],
})
continue
hits_out.append({
"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,
"section_filter": section_filter or None,
"total_drugs": total,
"total_drugs": len({h["yj_full"] for h in hits_out}),
"shown": len(hits_out),
"hits": hits_out,
}
if total > len(hits_out):
out["_more_count"] = total - len(hits_out)
return _dump(out)
"hits": hits_out[:limit],
})
def _tool_list_drug_chapters(yj_full: str) -> str:
"""List all chapter titles for a drug's package insert."""
client = _get_os()
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:
sections = MOCK_CHAPTERS.get(yj_full)
if not sections:
return _dump({"error": f"yj_full {yj_full} の章節が見つかりません。"})
sections = []
for h in hits:
src = h.get("_source") or {}
# Calculate text length from _score or use stored field
sections.append({
"section_title": src.get("section_title", ""),
"line_num": src.get("line_num", 0),
"text_len": 0, # not available from list query
})
drug = None
for d in MOCK_DRUG_MASTER.values():
if d["yj_full"] == yj_full:
drug = d
break
head = hits[0].get("_source") or {}
return _dump({
"yj_full": yj_full,
"brand": (head.get("brand_names") or [""])[0],
"generic": head.get("generic_name", ""),
"brand": drug["brand_name"] if drug else "",
"generic": drug["generic_name"] if drug else "",
"n_sections": len(sections),
"sections": sections,
})
def _tool_read_drug_chapter(yj_full: str, section_title: str) -> str:
"""Read verbatim text of a specific chapter."""
client = _get_os()
body = {
"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
text = MOCK_SECTION_TEXT.get((yj_full, section_title))
if text:
return text[:8000]
return _dump({
"error": f"section_title {section_title!r} は {yj_full} に存在しません。",
"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
# ---------------------------------------------------------------------------
# Map tool names to their implementation functions
_TOOL_DISPATCH = {
"search_drugs": lambda args: _tool_search_drugs(
query=args.get("query", ""),