增加子agent支持

agent/deep_assistant.py

@@ -43,6 +43,7 @@ from .mem0_middleware import create_mem0_middleware
 from .mem0_config import Mem0Config
 from agent.prompt_loader import load_system_prompt_async, load_mcp_settings_async
 from agent.agent_memory_cache import get_memory_cache_manager
+from .subagent_loader import load_subagents
 from .checkpoint_manager import get_checkpointer_manager
 from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
 from langgraph.checkpoint.memory import InMemorySaver
@@ -64,6 +65,8 @@ from deepagents.graph import BASE_AGENT_PROMPT
 from deepagents_cli.local_context import LocalContextMiddleware
 # Custom: FilesystemMiddleware with full SKILL.md reading support
 from .custom_filesystem_middleware import CustomFilesystemMiddleware
+# Sub-agent support
+from deepagents.middleware.subagents import SubAgent, SubAgentMiddleware
 
 # Global MemorySaver instance
 # from langgraph.checkpoint.memory import MemorySaver
@@ -308,6 +311,15 @@ async def init_agent(config: AgentConfig):
     sandbox, sandbox_type, workspace_root = await sandbox_task
     logger.info(f"init_agent sandbox ready, elapsed: {time.time() - create_start:.3f}s")
 
+    # Load sub-agents from skill directories
+    subagents = await load_subagents(
+        bot_id=config.bot_id,
+        tools=mcp_tools,
+        model=llm_instance,
+    )
+    if subagents:
+        logger.info(f"Loaded {len(subagents)} sub-agents: {[s['name'] for s in subagents]}")
+
     agent, composite_backend = create_custom_cli_agent(
         model=llm_instance,
         assistant_id=config.bot_id,
@@ -319,6 +331,7 @@ async def init_agent(config: AgentConfig):
         checkpointer=checkpointer,
         sandbox=sandbox,
         sandbox_type=sandbox_type,
+        subagents=subagents if subagents else None,
         shell_env={
             "ASSISTANT_ID": config.bot_id,
             "USER_IDENTIFIER": config.user_identifier,
@@ -385,6 +398,7 @@ def create_custom_cli_agent(
     checkpointer: Checkpointer | None = None,
     store: BaseStore | None = None,
     shell_env: dict[str, str] | None = None,
+    subagents: list[SubAgent] | None = None,
 ) -> tuple[Pregel, CompositeBackend]:
     """Create a CLI-configured agent with custom workspace_root for shell commands.
 
@@ -521,9 +535,19 @@ def create_custom_cli_agent(
         TodoListMiddleware(),
         FilePathFixMiddleware(),  # Fix extra spaces in CJK file names within tool call arguments
         CustomFilesystemMiddleware(backend=composite_backend),  # Use the custom FilesystemMiddleware with full SKILL.md reading support
+    ]
+    # Insert SubAgentMiddleware after FilesystemMiddleware (matches create_deep_agent ordering)
+    if subagents:
+        subagent_middleware = SubAgentMiddleware(
+            backend=composite_backend,
+            subagents=subagents,
+        )
+        deepagent_middleware.append(subagent_middleware)
+        logger.info(f"SubAgentMiddleware added with {len(subagents)} sub-agents: {[s['name'] for s in subagents]}")
+    deepagent_middleware.extend([
         AnthropicPromptCachingMiddleware(unsupported_model_behavior="ignore"),
         PatchToolCallsMiddleware(),
-    ]
+    ])
     if agent_middleware:
         deepagent_middleware.extend(agent_middleware)
     if interrupt_on is not None:

agent/subagent_loader.py

@@ -0,0 +1,188 @@
+"""Sub-agent loader for discovering and parsing sub-agent definitions from skill directories.
+
+Sub-agents are defined as markdown files with YAML frontmatter in skill directories:
+    projects/robot/{bot_id}/skills/{skill_name}/agents/*.md
+
+Each file has the format:
+    ---
+    name: code-reviewer
+    description: Reviews code for quality and security issues.
+    tools: rag_retrieve, table_rag_retrieve
+    ---
+
+    System prompt for the sub-agent...
+"""
+
+import logging
+import os
+import re
+from pathlib import Path
+from typing import Optional
+
+import yaml
+from deepagents.middleware.subagents import SubAgent
+from langchain.tools import BaseTool
+from langchain_core.language_models import BaseChatModel
+
+from agent.plugin_hook_loader import _get_skill_dirs
+
+logger = logging.getLogger('app')
+
+# Regex to extract YAML frontmatter and body from markdown files
+_FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n?(.*)$", re.DOTALL)
+
+
+def _parse_agent_md(file_path: Path) -> Optional[dict]:
+    """Parse a sub-agent markdown file with YAML frontmatter.
+
+    Args:
+        file_path: Path to the .md file.
+
+    Returns:
+        Dict with keys: name, description, system_prompt, tool_names (list[str] | None).
+        None if parsing fails.
+    """
+    try:
+        content = file_path.read_text(encoding="utf-8")
+    except OSError as e:
+        logger.warning(f"Failed to read sub-agent file {file_path}: {e}")
+        return None
+
+    match = _FRONTMATTER_RE.match(content)
+    if not match:
+        logger.warning(f"Sub-agent file {file_path} has no valid frontmatter")
+        return None
+
+    frontmatter_str, body = match.group(1), match.group(2)
+
+    try:
+        frontmatter = yaml.safe_load(frontmatter_str)
+    except yaml.YAMLError as e:
+        logger.warning(f"Invalid YAML in sub-agent file {file_path}: {e}")
+        return None
+
+    if not isinstance(frontmatter, dict):
+        logger.warning(f"Frontmatter in {file_path} is not a dict")
+        return None
+
+    name = frontmatter.get("name", "").strip() if isinstance(frontmatter.get("name"), str) else ""
+    description = frontmatter.get("description", "").strip() if isinstance(frontmatter.get("description"), str) else ""
+
+    if not name:
+        logger.warning(f"Sub-agent file {file_path} missing required 'name' field")
+        return None
+    if not description:
+        logger.warning(f"Sub-agent file {file_path} missing required 'description' field")
+        return None
+
+    # Parse optional tools field: comma-separated tool names
+    tool_names = None
+    tools_field = frontmatter.get("tools")
+    if tools_field is not None:
+        if isinstance(tools_field, str):
+            tool_names = [t.strip() for t in tools_field.split(",") if t.strip()]
+        elif isinstance(tools_field, list):
+            tool_names = [str(t).strip() for t in tools_field if str(t).strip()]
+        else:
+            logger.warning(f"Invalid 'tools' field in {file_path}, expected string or list")
+
+    return {
+        "name": name,
+        "description": description,
+        "system_prompt": body.strip(),
+        "tool_names": tool_names,
+        "source": str(file_path),
+    }
+
+
+def _filter_tools_by_names(all_tools: list[BaseTool], tool_names: list[str]) -> list[BaseTool]:
+    """Filter MCP tools by name whitelist.
+
+    Args:
+        all_tools: All available MCP tools.
+        tool_names: Whitelist of tool names to include.
+
+    Returns:
+        Filtered list of tools. Logs warning for names not found.
+    """
+    tool_lookup = {tool.name: tool for tool in all_tools}
+    filtered = []
+    for name in tool_names:
+        if name in tool_lookup:
+            filtered.append(tool_lookup[name])
+        else:
+            available = list(tool_lookup.keys())
+            logger.warning(f"Sub-agent tool '{name}' not found in MCP tools. Available: {available}")
+    return filtered
+
+
+async def load_subagents(
+    bot_id: str,
+    tools: list[BaseTool],
+    model: BaseChatModel,
+) -> list[SubAgent]:
+    """Load sub-agent definitions from skill directories.
+
+    Scans all skill directories for the given bot_id, looking for agents/*.md files
+    in each skill subdirectory.
+
+    Args:
+        bot_id: Bot identifier for locating skill directories.
+        tools: All available MCP tools for filtering.
+        model: The main agent's model, used by each sub-agent.
+
+    Returns:
+        List of SubAgent dicts. Empty list if no sub-agents found.
+    """
+    skill_dirs = _get_skill_dirs(bot_id)
+    parsed_agents: dict[str, dict] = {}  # name -> parsed dict (last-wins for dedup)
+
+    for skill_dir in skill_dirs:
+        if not os.path.exists(skill_dir):
+            continue
+
+        for skill_name in os.listdir(skill_dir):
+            skill_path = os.path.join(skill_dir, skill_name)
+            if not os.path.isdir(skill_path):
+                continue
+
+            agents_dir = Path(skill_path) / "agents"
+            if not agents_dir.exists():
+                continue
+
+            for md_file in agents_dir.glob("*.md"):
+                parsed = _parse_agent_md(md_file)
+                if parsed is None:
+                    continue
+
+                name = parsed["name"]
+                if name in parsed_agents:
+                    logger.warning(
+                        f"Duplicate sub-agent name '{name}': "
+                        f"{parsed_agents[name]['source']} overridden by {parsed['source']}"
+                    )
+                parsed_agents[name] = parsed
+
+    if not parsed_agents:
+        return []
+
+    # Build SubAgent dicts with model and filtered tools
+    subagents: list[SubAgent] = []
+    for name, parsed in parsed_agents.items():
+        # Filter tools: if tool_names specified, filter; otherwise inherit all
+        if parsed["tool_names"] is not None:
+            filtered_tools = _filter_tools_by_names(tools, parsed["tool_names"])
+        else:
+            filtered_tools = list(tools)
+
+        subagent: SubAgent = {
+            "name": name,
+            "description": parsed["description"],
+            "system_prompt": parsed["system_prompt"],
+            "model": model,
+            "tools": filtered_tools,
+        }
+        subagents.append(subagent)
+        logger.info(f"Loaded sub-agent '{name}' with {len(filtered_tools)} tools from {parsed['source']}")
+
+    return subagents

skills/developing/pmda-drug-info/.claude-plugin/plugin.json

@@ -0,0 +1,21 @@
+{
+  "name": "pmda-drug-info",
+  "description": "PMDA drug information tools for Japanese pharmaceutical package insert queries. Provides drug search, master info, interactions, restrictions, dosing, and full-text chapter retrieval via PostgreSQL + OpenSearch.",
+  "hooks": {
+    "PrePrompt": [
+      {
+        "type": "command",
+        "command": "python hooks/pre_prompt.py"
+      }
+    ]
+  },
+  "mcpServers": {
+    "pmda_drug_info": {
+      "transport": "stdio",
+      "command": "python",
+      "args": [
+        "./pmda_server.py"
+      ]
+    }
+  }
+}

skills/developing/pmda-drug-info/agents/adverse-event.md

@@ -0,0 +1,31 @@
+---
+name: adverse_event
+description: Reverse lookup drugs by adverse event name. Find which drugs have reported a specific side effect.
+             Invoke when the user asks "Which drugs cause Stevens-Johnson syndrome?" or "Drugs that prolong QT interval?".
+             Causal inference is prohibited — information presentation only.
+tools: search_section_text, search_drugs, get_drug_master, list_drug_chapters, read_drug_chapter
+---
+
+あなたは「副作用 → 該当薬剤の逆引き」専門の sub-agent です。
+
+【ツール戦略】
+1. `search_section_text(keyword=副作用名, section_filter="副作用")` で逆引き。
+   total_drugs は必ず本文中に明示する。
+2. 同義語が必要なケース:
+     "Stevens-Johnson" ⇔ "皮膚粘膜眼症候群" / "SJS"
+     "QT延長" ⇔ "Torsades de pointes"
+     "間質性肺炎" ⇔ "肺臓炎"
+   OS の synonym filter が自動展開するので 1 回の検索で OK。
+3. hit から代表薬を 3〜5 件選び、`read_drug_chapter` で 11.1 重大な副作用 / 11.2 その他の副作用
+   verbatim を引用。
+4. 因果推論（"この薬がこの患者の症状を起こした"）は **絶対しない**。
+   情報提示のみ。
+
+【絶対ルール】
+1. ツール呼び出し必須。トレーニング知識・教科書・ガイドラインからの推測は禁止。
+2. 数値・固有名・条件は本文表現を改変せず逐語引用。
+3. 出典は **必ず** `[出典: <販売名> (yj_full=<id>) / <章番号 章タイトル>]` の形式。
+   - fact 表 row には `_citation` フィールドが入っているので **そのまま転記**。
+   - `[出典: 薬品マスター]` `[出典: 添付文書]` 等の汎用出典は **絶対禁止**。
+   - read_drug_chapter で実際に読んだ section 以外の出典を捏造しない。
+4. 該当情報が無ければ "添付文書からは確認できません" と書く。

skills/developing/pmda-drug-info/agents/interaction.md

@@ -0,0 +1,28 @@
+---
+name: interaction
+description: Investigate drug-drug interactions between two drugs, or list all interactions for a single drug.
+             Invoke when the user asks "Can drug A and B be used together?" or "What are the interactions of drug A?".
+tools: search_drugs, get_drug_master, get_drug_interactions, search_section_text, list_drug_chapters, read_drug_chapter
+---
+
+あなたは「薬剤間相互作用」専門の sub-agent です。
+
+【ツール戦略】
+- A・B 両薬の yj_code を `search_drugs` で取得。
+- `get_drug_interactions(drug_a_yj=A, drug_b_yj=B)` で双方向検索（A→B も B→A も拾える）。
+- ヒットしたら drug_a の側の出典 section（10.1 / 10.2）を `list_drug_chapters` + `read_drug_chapter` で
+  verbatim 取得。drug_b 側にも該当記載があるか確認。
+- ヒットゼロ → "添付文書上は併用禁忌・併用注意の明確な記載なし" と書く（自由記述/警告等は
+  別途 `search_section_text(keyword=B薬名, section_filter="相互作用")` で念押し）。
+- 1 薬名のみ与えられた場合は `get_drug_interactions(drug_a_yj=...)` で全相互作用一覧。
+
+severity は本文の "併用禁忌" / "併用注意" の語をそのまま転記。
+
+【絶対ルール】
+1. ツール呼び出し必須。トレーニング知識・教科書・ガイドラインからの推測は禁止。
+2. 数値・固有名・条件は本文表現を改変せず逐語引用。
+3. 出典は **必ず** `[出典: <販売名> (yj_full=<id>) / <章番号 章タイトル>]` の形式。
+   - fact 表 row には `_citation` フィールドが入っているので **そのまま転記**。
+   - `[出典: 薬品マスター]` `[出典: 添付文書]` 等の汎用出典は **絶対禁止**。
+   - read_drug_chapter で実際に読んだ section 以外の出典を捏造しない。
+4. 該当情報が無ければ "添付文書からは確認できません" と書く。

skills/developing/pmda-drug-info/agents/patient-specific.md

@@ -0,0 +1,32 @@
+---
+name: patient_specific
+description: Determine drug administration feasibility and dosage adjustment for specific patient conditions (renal impairment, hepatic impairment, pregnancy, elderly, pediatric, allergy).
+             Invoke when the user asks "Can this drug be used in a patient with eGFR 25?", "Is it contraindicated in pregnancy?", etc.
+tools: search_drugs, get_drug_master, get_drug_restrictions, get_drug_dosing, list_drug_chapters, read_drug_chapter
+---
+
+あなたは「特定患者への投与可否・用量調整」専門の sub-agent です。
+
+【ツール戦略】
+1. 薬名から yj_code を `search_drugs` で取得。
+2. 患者条件を condition_type に対応付け:
+     - 腎機能 (eGFR/CrCl) → "腎機能障害"
+     - 肝機能 (Child-Pugh) → "肝機能障害"
+     - 妊娠/授乳 → "妊婦"/"授乳婦"
+     - 年齢 (小児/高齢) → "小児等"/"高齢者"
+     - アレルギー既往 → "過敏症"
+     - 合併症 (糖尿病/喘息など) → "疾患"
+3. `get_drug_restrictions(drug_yj=..., condition_type=...)` で該当 restriction を取得。
+   condition_params の数値（例: {"eGFR_max": 30}）を必ず確認。
+4. `get_drug_dosing(drug_yj=..., patient_segment=...)` で患者層別用量を取得。
+5. 必要なら原文 `read_drug_chapter` で 9.x 章 verbatim 引用。
+6. 数値判定（例: eGFR=25 ⇔ eGFR_max=30 → 該当）を agent が責任もって行う。
+
+【絶対ルール】
+1. ツール呼び出し必須。トレーニング知識・教科書・ガイドラインからの推測は禁止。
+2. 数値・固有名・条件は本文表現を改変せず逐語引用。
+3. 出典は **必ず** `[出典: <販売名> (yj_full=<id>) / <章番号 章タイトル>]` の形式。
+   - fact 表 row には `_citation` フィールドが入っているので **そのまま転記**。
+   - `[出典: 薬品マスター]` `[出典: 添付文書]` 等の汎用出典は **絶対禁止**。
+   - read_drug_chapter で実際に読んだ section 以外の出典を捏造しない。
+4. 該当情報が無ければ "添付文書からは確認できません" と書く。

skills/developing/pmda-drug-info/agents/single-drug.md

@@ -0,0 +1,26 @@
+---
+name: single_drug
+description: Answer factual questions about a single drug (brand name, generic name, indications, dosing, contraindications, side effects, etc.).
+             Invoke when the question is focused on one drug and requires detailed information from the package insert.
+tools: search_drugs, get_drug_master, get_drug_dosing, get_drug_restrictions, list_drug_chapters, read_drug_chapter
+---
+
+あなたは「単一薬の事実回答」専門の sub-agent です。
+
+【ツール戦略】
+1. 質問から薬名/yj_code を特定 → `search_drugs` または直接 yj_code が分かれば次へ。
+2. `get_drug_master(yj_code)` で基本情報（販売名・一般名・薬効分類・規制）を確定。
+3. 必要に応じて `get_drug_dosing` で用法用量、`get_drug_restrictions(drug_yj=...)` で禁忌・特定患者注意。
+4. 自由記述や上記テーブルに無い情報（例: 重大な副作用一覧、薬物動態の数値）は
+   `list_drug_chapters(yj_full)` → `read_drug_chapter(yj_full, section_title)` で原文取得。
+
+最終回答は箇条書き or 表で、各事実に出典を付ける。
+
+【絶対ルール】
+1. ツール呼び出し必須。トレーニング知識・教科書・ガイドラインからの推測は禁止。
+2. 数値・固有名・条件は本文表現を改変せず逐語引用。
+3. 出典は **必ず** `[出典: <販売名> (yj_full=<id>) / <章番号 章タイトル>]` の形式。
+   - fact 表 row には `_citation` フィールドが入っているので **そのまま転記**。
+   - `[出典: 薬品マスター]` `[出典: 添付文書]` 等の汎用出典は **絶対禁止**。
+   - read_drug_chapter で実際に読んだ section 以外の出典を捏造しない。
+4. 該当情報が無ければ "添付文書からは確認できません" と書く。

skills/developing/pmda-drug-info/hooks/pmda-instructions.md

@@ -0,0 +1,22 @@
+# PMDA Drug Information Tools
+
+You have access to Japanese pharmaceutical package insert (添付文書) data via the following tools.
+
+## Core Rules
+- **Tool calls are mandatory.** Never answer from training knowledge alone. All facts must come from tool results.
+- Cite sources in the format: `[出典: <販売名> (yj_full=<id>) / <章番号 章タイトル>]`
+- Fact table rows include a `_citation` field — use it directly.
+- Generic citations like `[出典: 薬品マスター]` or `[出典: 添付文書]` are **prohibited**.
+- For urgent questions (suicide/drug abuse/severe acute symptoms), state: "緊急対応として担当医・薬剤師に直接相談してください"
+
+## When to Use Sub-agents (task tool)
+- **patient_specific**: Renal/hepatic/pregnancy/elderly/pediatric/allergy conditions × dosing decisions
+- **interaction**: Pairwise drug interaction investigation
+- **adverse_event**: Reverse lookup from adverse event name to drugs
+- **single_drug**: Detailed info not in fact tables (e.g., full adverse event list, pharmacokinetics)
+
+## Direct Tool Usage (do NOT delegate)
+- Simple lookups → use tools directly
+- Multi-drug comparisons → call tools sequentially, output as markdown table
+- Symptom → candidate drug reverse lookup → `search_section_text`
+- Mechanism/pharmacokinetics → `list_drug_chapters` + `read_drug_chapter`

skills/developing/pmda-drug-info/hooks/pre_prompt.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python3
+"""
+PrePrompt hook for PMDA drug info skill.
+Injects usage instructions for the drug information tools.
+"""
+import sys
+from pathlib import Path
+
+
+def main():
+    prompt_file = Path(__file__).parent / "pmda-instructions.md"
+    if prompt_file.exists():
+        print(prompt_file.read_text(encoding="utf-8"))
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

skills/developing/pmda-drug-info/mcp_common.py

@@ -0,0 +1,252 @@
+#!/usr/bin/env python3
+"""
+Shared utility functions for the MCP server.
+Provides common functionality for path handling, file validation, and request processing.
+"""
+
+import json
+import os
+import sys
+import asyncio
+from typing import Any, Dict, List, Optional, Union
+import re
+
+def get_allowed_directory():
+    """Get the directory that is allowed to be accessed."""
+    # Prefer dataset_dir passed through command-line arguments.
+    if len(sys.argv) > 1:
+        dataset_dir = sys.argv[1]
+        return os.path.abspath(dataset_dir)
+
+    # Read the project data directory from the environment variable.
+    project_dir = os.getenv("PROJECT_DATA_DIR", "./projects/data")
+    return os.path.abspath(project_dir)
+
+
+def resolve_file_path(file_path: str, default_subfolder: str = "default") -> str:
+    """
+    Resolve a file path, supporting both folder/document.txt and document.txt formats.
+
+    Args:
+        file_path: Input file path.
+        default_subfolder: Default subfolder name to use when only a filename is provided.
+
+    Returns:
+        The resolved full file path.
+    """
+    # If the path contains a folder separator, use it directly.
+    if '/' in file_path or '\\' in file_path:
+        clean_path = file_path.replace('\\', '/')
+
+        # Remove the projects/ prefix if it exists.
+        if clean_path.startswith('projects/'):
+            clean_path = clean_path[9:]  # Remove the 'projects/' prefix.
+        elif clean_path.startswith('./projects/'):
+            clean_path = clean_path[11:]  # Remove the './projects/' prefix.
+    else:
+        # If only a filename is provided, add the default subfolder.
+        clean_path = f"{default_subfolder}/{file_path}"
+
+    # Get the allowed directory.
+    project_data_dir = get_allowed_directory()
+
+    # Try to locate the file directly under the project directory.
+    full_path = os.path.join(project_data_dir, clean_path.lstrip('./'))
+    if os.path.exists(full_path):
+        return full_path
+
+    # If the direct path does not exist, try a recursive search.
+    found = find_file_in_project(clean_path, project_data_dir)
+    if found:
+        return found
+
+    # If this is a bare filename and it was not found under the default subfolder,
+    # try looking in the project root.
+    if '/' not in file_path and '\\' not in file_path:
+        root_path = os.path.join(project_data_dir, file_path)
+        if os.path.exists(root_path):
+            return root_path
+
+    raise FileNotFoundError(f"File not found: {file_path} (searched in {project_data_dir})")
+
+
+def find_file_in_project(filename: str, project_dir: str) -> Optional[str]:
+    """Recursively search for a file inside the project directory."""
+    # If filename includes a path, only search within the specified path.
+    if '/' in filename:
+        parts = filename.split('/')
+        target_file = parts[-1]
+        search_dir = os.path.join(project_dir, *parts[:-1])
+
+        if os.path.exists(search_dir):
+            target_path = os.path.join(search_dir, target_file)
+            if os.path.exists(target_path):
+                return target_path
+    else:
+        # For a bare filename, recursively search the whole project directory.
+        for root, dirs, files in os.walk(project_dir):
+            if filename in files:
+                return os.path.join(root, filename)
+    return None
+
+
+def load_tools_from_json(tools_file_name: str) -> List[Dict[str, Any]]:
+    """Load tool definitions from a JSON file."""
+    try:
+        tools_file = os.path.join(os.path.dirname(__file__), tools_file_name)
+        if os.path.exists(tools_file):
+            with open(tools_file, 'r', encoding='utf-8') as f:
+                return json.load(f)
+        else:
+            # If the JSON file does not exist, use the default definitions.
+            return []
+    except Exception as e:
+        print(f"Warning: Unable to load tool definition JSON file: {str(e)}")
+        return []
+
+
+def create_error_response(request_id: Any, code: int, message: str) -> Dict[str, Any]:
+    """Create a standardized error response."""
+    return {
+        "jsonrpc": "2.0",
+        "id": request_id,
+        "error": {
+            "code": code,
+            "message": message
+        }
+    }
+
+
+def create_success_response(request_id: Any, result: Any) -> Dict[str, Any]:
+    """Create a standardized success response."""
+    return {
+        "jsonrpc": "2.0",
+        "id": request_id,
+        "result": result
+    }
+
+
+def create_initialize_response(request_id: Any, server_name: str, server_version: str = "1.0.0") -> Dict[str, Any]:
+    """Create a standardized initialize response."""
+    return {
+        "jsonrpc": "2.0",
+        "id": request_id,
+        "result": {
+            "protocolVersion": "2024-11-05",
+            "capabilities": {
+                "tools": {}
+            },
+            "serverInfo": {
+                "name": server_name,
+                "version": server_version
+            }
+        }
+    }
+
+
+def create_ping_response(request_id: Any) -> Dict[str, Any]:
+    """Create a standardized ping response."""
+    return {
+        "jsonrpc": "2.0",
+        "id": request_id,
+        "result": {
+            "pong": True
+        }
+    }
+
+
+def create_tools_list_response(request_id: Any, tools: List[Dict[str, Any]]) -> Dict[str, Any]:
+    """Create a standardized tools/list response."""
+    return {
+        "jsonrpc": "2.0",
+        "id": request_id,
+        "result": {
+            "tools": tools
+        }
+    }
+
+
+def is_regex_pattern(pattern: str) -> bool:
+    """Check whether a string should be treated as a regular expression pattern."""
+    # Check the /pattern/ format.
+    if pattern.startswith('/') and pattern.endswith('/') and len(pattern) > 2:
+        return True
+
+    # Check the r"pattern" or r'pattern' format.
+    if pattern.startswith(('r"', "r'")) and pattern.endswith(('"', "'")) and len(pattern) > 3:
+        return True
+
+    # Check whether it contains regex metacharacters.
+    regex_chars = {'*', '+', '?', '|', '(', ')', '[', ']', '{', '}', '^', '$', '\\', '.'}
+    return any(char in pattern for char in regex_chars)
+
+
+def compile_pattern(pattern: str) -> Union[re.Pattern, str, None]:
+    """Compile a regex pattern, or return the original string if it is not regex."""
+    if not is_regex_pattern(pattern):
+        return pattern
+
+    try:
+        # Handle the /pattern/ format.
+        if pattern.startswith('/') and pattern.endswith('/'):
+            regex_body = pattern[1:-1]
+            return re.compile(regex_body)
+
+        # Handle the r"pattern" or r'pattern' format.
+        if pattern.startswith(('r"', "r'")) and pattern.endswith(('"', "'")):
+            regex_body = pattern[2:-1]
+            return re.compile(regex_body)
+
+        # Directly compile strings that contain regex metacharacters.
+        return re.compile(pattern)
+    except re.error as e:
+        # If compilation fails, return None to indicate an invalid regex.
+        print(f"Warning: Regular expression '{pattern}' compilation failed: {e}")
+        return None
+
+
+async def handle_mcp_streaming(request_handler):
+    """Handle the standard main loop for MCP requests."""
+    try:
+        while True:
+            # Read from stdin
+            line = await asyncio.get_event_loop().run_in_executor(None, sys.stdin.readline)
+            if not line:
+                break
+
+            line = line.strip()
+            if not line:
+                continue
+
+            try:
+                request = json.loads(line)
+                response = await request_handler(request)
+
+                # Write to stdout
+                sys.stdout.write(json.dumps(response, ensure_ascii=False) + "\n")
+                sys.stdout.flush()
+
+            except json.JSONDecodeError:
+                error_response = {
+                    "jsonrpc": "2.0",
+                    "error": {
+                        "code": -32700,
+                        "message": "Parse error"
+                    }
+                }
+                sys.stdout.write(json.dumps(error_response, ensure_ascii=False) + "\n")
+                sys.stdout.flush()
+
+            except Exception as e:
+                error_response = {
+                    "jsonrpc": "2.0",
+                    "error": {
+                        "code": -32603,
+                        "message": f"Internal error: {str(e)}"
+                    }
+                }
+                sys.stdout.write(json.dumps(error_response, ensure_ascii=False) + "\n")
+                sys.stdout.flush()
+
+    except KeyboardInterrupt:
+        pass

skills/developing/pmda-drug-info/pmda_server.py

@@ -0,0 +1,596 @@
+#!/usr/bin/env python3
+"""
+PMDA drug information MCP server.
+Provides drug search, master info, interactions, restrictions, dosing,
+and full-text chapter retrieval via PostgreSQL + OpenSearch.
+"""
+
+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 mcp_common import (
+    create_error_response,
+    create_initialize_response,
+    create_ping_response,
+    create_tools_list_response,
+    load_tools_from_json,
+    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)
+
+
+# ---------------------------------------------------------------------------
+# Lazy database connections
+# ---------------------------------------------------------------------------
+_pg_conn = None
+_os_client = None
+
+# Drug lookup cache: yj_code -> (brand_name, yj_full)
+_drug_lookup: Optional[Dict[str, tuple]] = None
+
+
+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
+
+
+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
+
+
+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
+
+
+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))
+    chap = section or "（章不明）"
+    return f"[出典: {brand} (yj_full={yj_full}) / {chap}]"
+
+
+# ---------------------------------------------------------------------------
+# Tool implementations
+# ---------------------------------------------------------------------------
+
+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
+    ])
+
+
+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])
+
+
+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])
+
+
+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:
+        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)
+
+
+def _tool_get_drug_interactions(
+    drug_a_yj: Optional[str] = None,
+    drug_b_yj: Optional[str] = None,
+    severity: Optional[str] = None,
+    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
+    ])
+
+
+def _tool_get_drug_restrictions(
+    drug_yj: Optional[str] = None,
+    condition_type: Optional[str] = None,
+    severity: Optional[str] = None,
+    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
+    ])
+
+
+def _tool_get_drug_dosing(
+    drug_yj: str,
+    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
+    ])
+
+
+def _tool_search_section_text(
+    keyword: str,
+    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"])
+
+    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:
+                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}) / <該当章>]",
+        })
+
+    out = {
+        "keyword": keyword,
+        "section_filter": section_filter or None,
+        "total_drugs": total,
+        "shown": len(hits_out),
+        "hits": hits_out,
+    }
+    if total > len(hits_out):
+        out["_more_count"] = total - len(hits_out)
+    return _dump(out)
+
+
+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:
+        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
+        })
+
+    head = hits[0].get("_source") or {}
+    return _dump({
+        "yj_full": yj_full,
+        "brand": (head.get("brand_names") or [""])[0],
+        "generic": head.get("generic_name", ""),
+        "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
+    return _dump({
+        "error": f"section_title {section_title!r} は {yj_full} に存在しません。",
+        "hint": "list_drug_chapters で取得した sections[].section_title をそのまま渡してください。",
+    })
+
+
+# ---------------------------------------------------------------------------
+# MCP request handler
+# ---------------------------------------------------------------------------
+
+# Map tool names to their implementation functions
+_TOOL_DISPATCH = {
+    "search_drugs": lambda args: _tool_search_drugs(
+        query=args.get("query", ""),
+        kind=args.get("kind", "auto"),
+        limit=args.get("limit", 10),
+    ),
+    "list_categories": lambda args: _tool_list_categories(),
+    "list_drugs_in_category": lambda args: _tool_list_drugs_in_category(
+        l2_code=args.get("l2_code", ""),
+        limit_generics=args.get("limit_generics", 50),
+    ),
+    "get_drug_master": lambda args: _tool_get_drug_master(
+        yj_code=args.get("yj_code", ""),
+    ),
+    "get_drug_interactions": lambda args: _tool_get_drug_interactions(
+        drug_a_yj=args.get("drug_a_yj"),
+        drug_b_yj=args.get("drug_b_yj"),
+        severity=args.get("severity"),
+        keyword=args.get("keyword"),
+        limit=args.get("limit", 30),
+    ),
+    "get_drug_restrictions": lambda args: _tool_get_drug_restrictions(
+        drug_yj=args.get("drug_yj"),
+        condition_type=args.get("condition_type"),
+        severity=args.get("severity"),
+        keyword=args.get("keyword"),
+        limit=args.get("limit", 30),
+    ),
+    "get_drug_dosing": lambda args: _tool_get_drug_dosing(
+        drug_yj=args.get("drug_yj", ""),
+        patient_segment=args.get("patient_segment"),
+        limit=args.get("limit", 20),
+    ),
+    "search_section_text": lambda args: _tool_search_section_text(
+        keyword=args.get("keyword", ""),
+        section_filter=args.get("section_filter", ""),
+        limit=args.get("limit", 30),
+    ),
+    "list_drug_chapters": lambda args: _tool_list_drug_chapters(
+        yj_full=args.get("yj_full", ""),
+    ),
+    "read_drug_chapter": lambda args: _tool_read_drug_chapter(
+        yj_full=args.get("yj_full", ""),
+        section_title=args.get("section_title", ""),
+    ),
+}
+
+
+async def handle_request(request: Dict[str, Any]) -> Dict[str, Any]:
+    """Handle an MCP request."""
+    try:
+        method = request.get("method")
+        params = request.get("params", {})
+        request_id = request.get("id")
+
+        if method == "initialize":
+            return create_initialize_response(request_id, "pmda-drug-info")
+
+        elif method == "ping":
+            return create_ping_response(request_id)
+
+        elif method == "tools/list":
+            tools = load_tools_from_json("pmda_tools.json")
+            return create_tools_list_response(request_id, tools)
+
+        elif method == "tools/call":
+            tool_name = params.get("name")
+            arguments = params.get("arguments", {})
+
+            if tool_name not in _TOOL_DISPATCH:
+                return create_error_response(request_id, -32601, f"Unknown tool: {tool_name}")
+
+            try:
+                result_text = _TOOL_DISPATCH[tool_name](arguments)
+                return {
+                    "jsonrpc": "2.0",
+                    "id": request_id,
+                    "result": {
+                        "content": [{"type": "text", "text": result_text}]
+                    },
+                }
+            except Exception as e:
+                return {
+                    "jsonrpc": "2.0",
+                    "id": request_id,
+                    "result": {
+                        "content": [{"type": "text", "text": f"Error: {str(e)}"}]
+                    },
+                }
+
+        else:
+            return create_error_response(request_id, -32601, f"Unknown method: {method}")
+
+    except Exception as e:
+        return create_error_response(request.get("id"), -32603, f"Internal error: {str(e)}")
+
+
+async def main():
+    await handle_mcp_streaming(handle_request)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

skills/developing/pmda-drug-info/pmda_tools.json

@@ -0,0 +1,207 @@
+[
+  {
+    "name": "search_drugs",
+    "description": "Search drugs by brand name, generic name, or YJ code. Returns list of matching drugs with yj_code, brand name, generic name, and category.",
+    "inputSchema": {
+      "type": "object",
+      "properties": {
+        "query": {
+          "type": "string",
+          "description": "Search query: drug brand name, generic name, or YJ code."
+        },
+        "kind": {
+          "type": "string",
+          "enum": ["auto", "brand", "generic", "yj"],
+          "description": "Search type. 'auto' searches all fields.",
+          "default": "auto"
+        },
+        "limit": {
+          "type": "integer",
+          "description": "Maximum number of results.",
+          "default": 10
+        }
+      },
+      "required": ["query"]
+    }
+  },
+  {
+    "name": "list_categories",
+    "description": "List all L1/L2 drug categories (pharmacological classification) with drug counts per category.",
+    "inputSchema": {
+      "type": "object",
+      "properties": {}
+    }
+  },
+  {
+    "name": "list_drugs_in_category",
+    "description": "List all drugs (generic → brand names) under a specific L2 pharmacological category code.",
+    "inputSchema": {
+      "type": "object",
+      "properties": {
+        "l2_code": {
+          "type": "string",
+          "description": "3-digit L2 category code."
+        },
+        "limit_generics": {
+          "type": "integer",
+          "description": "Maximum number of generic names to return.",
+          "default": 50
+        }
+      },
+      "required": ["l2_code"]
+    }
+  },
+  {
+    "name": "get_drug_master",
+    "description": "Get basic information for a drug by yj_code: brand name, generic name, pharmacological category, regulatory classification, manufacturer, revision date.",
+    "inputSchema": {
+      "type": "object",
+      "properties": {
+        "yj_code": {
+          "type": "string",
+          "description": "12-character YJ code."
+        }
+      },
+      "required": ["yj_code"]
+    }
+  },
+  {
+    "name": "get_drug_interactions",
+    "description": "Search drug interactions. With drug_a only: all interactions for that drug. With both drug_a and drug_b: bidirectional interaction between A and B. Filter by severity (併用禁忌/併用注意) or keyword.",
+    "inputSchema": {
+      "type": "object",
+      "properties": {
+        "drug_a_yj": {
+          "type": "string",
+          "description": "YJ code for drug A."
+        },
+        "drug_b_yj": {
+          "type": "string",
+          "description": "YJ code for drug B (optional, for pairwise lookup)."
+        },
+        "severity": {
+          "type": "string",
+          "description": "Filter by severity: '併用禁忌' or '併用注意'."
+        },
+        "keyword": {
+          "type": "string",
+          "description": "Search keyword in drug_b_class, mechanism, or clinical_effect."
+        },
+        "limit": {
+          "type": "integer",
+          "description": "Maximum number of results.",
+          "default": 30
+        }
+      }
+    }
+  },
+  {
+    "name": "get_drug_restrictions",
+    "description": "Search drug restrictions (contraindications, precautions) by patient condition. condition_type options: 疾患, 腎機能障害, 肝機能障害, 生殖能, 妊婦, 授乳婦, 小児等, 高齢者, 過敏症, 遺伝子多型, その他. severity options: 禁忌, 原則禁忌, 慎重投与.",
+    "inputSchema": {
+      "type": "object",
+      "properties": {
+        "drug_yj": {
+          "type": "string",
+          "description": "YJ code for the drug."
+        },
+        "condition_type": {
+          "type": "string",
+          "description": "Patient condition type to filter by."
+        },
+        "severity": {
+          "type": "string",
+          "description": "Filter by severity: 禁忌, 原則禁忌, or 慎重投与."
+        },
+        "keyword": {
+          "type": "string",
+          "description": "Search keyword in condition_text."
+        },
+        "limit": {
+          "type": "integer",
+          "description": "Maximum number of results.",
+          "default": 30
+        }
+      }
+    }
+  },
+  {
+    "name": "get_drug_dosing",
+    "description": "Get dosing information for a drug, optionally filtered by patient segment. patient_segment options: 成人, 小児等, 高齢者, 腎機能障害患者, 肝機能障害患者, 透析患者, 妊婦.",
+    "inputSchema": {
+      "type": "object",
+      "properties": {
+        "drug_yj": {
+          "type": "string",
+          "description": "YJ code for the drug."
+        },
+        "patient_segment": {
+          "type": "string",
+          "description": "Patient segment to filter by (e.g., 成人, 高齢者, 腎機能障害患者)."
+        },
+        "limit": {
+          "type": "integer",
+          "description": "Maximum number of results.",
+          "default": 20
+        }
+      },
+      "required": ["drug_yj"]
+    }
+  },
+  {
+    "name": "search_section_text",
+    "description": "Full-text search in drug package insert sections. Returns matching sections with snippets. Use section_filter to narrow by chapter title (e.g., '副作用', '禁忌', '妊婦', '相互作用').",
+    "inputSchema": {
+      "type": "object",
+      "properties": {
+        "keyword": {
+          "type": "string",
+          "description": "Search keyword."
+        },
+        "section_filter": {
+          "type": "string",
+          "description": "Filter by section title substring (e.g., '副作用', '禁忌', '妊婦').",
+          "default": ""
+        },
+        "limit": {
+          "type": "integer",
+          "description": "Maximum number of results.",
+          "default": 30
+        }
+      },
+      "required": ["keyword"]
+    }
+  },
+  {
+    "name": "list_drug_chapters",
+    "description": "List all chapter titles for a drug's package insert. Use yj_full (full YJ code with revision suffix). Returns section titles with line numbers.",
+    "inputSchema": {
+      "type": "object",
+      "properties": {
+        "yj_full": {
+          "type": "string",
+          "description": "Full YJ code (with revision suffix, e.g., 3399007H1021_1_21)."
+        }
+      },
+      "required": ["yj_full"]
+    }
+  },
+  {
+    "name": "read_drug_chapter",
+    "description": "Read the verbatim text of a specific chapter from a drug's package insert. section_title must match exactly from list_drug_chapters output.",
+    "inputSchema": {
+      "type": "object",
+      "properties": {
+        "yj_full": {
+          "type": "string",
+          "description": "Full YJ code."
+        },
+        "section_title": {
+          "type": "string",
+          "description": "Exact section title from list_drug_chapters (e.g., '9.2 腎機能障害患者', '11.1 重大な副作用')."
+        }
+      },
+      "required": ["yj_full", "section_title"]
+    }
+  }
+]