feat: implement 7 Parlant tools

Add @tool decorated functions: extract_patient_profile,
generate_search_anchors, search_clinical_trials, refine_search_query,
relax_search_query, evaluate_trial_eligibility (dual-model), and
analyze_gaps. Each returns ToolResult with data and metadata.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

trialpath/agent/tools.py

@@ -0,0 +1,234 @@
+"""Parlant tool definitions for the TrialPath agent."""
+import json
+
+from parlant.sdk import ToolContext, ToolResult, tool
+
+from trialpath.config import (
+    GEMINI_API_KEY,
+    GEMINI_MODEL,
+    HF_TOKEN,
+    MCP_URL,
+    MEDGEMMA_ENDPOINT_URL,
+)
+
+
+@tool
+async def extract_patient_profile(
+    context: ToolContext,
+    document_urls: str,
+    metadata: str,
+) -> ToolResult:
+    """Extract a structured patient profile from uploaded medical documents.
+
+    Args:
+        context: Parlant tool context.
+        document_urls: JSON list of document file paths.
+        metadata: JSON object with known patient metadata (age, sex).
+    """
+    from trialpath.services.medgemma_extractor import MedGemmaExtractor
+
+    extractor = MedGemmaExtractor(
+        endpoint_url=MEDGEMMA_ENDPOINT_URL,
+        hf_token=HF_TOKEN,
+    )
+    urls = json.loads(document_urls)
+    meta = json.loads(metadata)
+    profile = await extractor.extract(urls, meta)
+
+    return ToolResult(
+        data=profile,
+        metadata={"source": "medgemma", "doc_count": len(urls)},
+    )
+
+
+@tool
+async def generate_search_anchors(
+    context: ToolContext,
+    patient_profile: str,
+) -> ToolResult:
+    """Generate search parameters from a patient profile for ClinicalTrials.gov.
+
+    Args:
+        context: Parlant tool context.
+        patient_profile: JSON string of PatientProfile data.
+    """
+    from trialpath.services.gemini_planner import GeminiPlanner
+
+    planner = GeminiPlanner(model=GEMINI_MODEL, api_key=GEMINI_API_KEY)
+    profile = json.loads(patient_profile)
+    anchors = await planner.generate_search_anchors(profile)
+
+    return ToolResult(
+        data=anchors.model_dump(),
+        metadata={"source": "gemini"},
+    )
+
+
+@tool
+async def search_clinical_trials(
+    context: ToolContext,
+    search_anchors: str,
+) -> ToolResult:
+    """Search ClinicalTrials.gov for matching trials using search anchors.
+
+    Args:
+        context: Parlant tool context.
+        search_anchors: JSON string of SearchAnchors data.
+    """
+    from trialpath.models.search_anchors import SearchAnchors
+    from trialpath.services.mcp_client import ClinicalTrialsMCPClient
+
+    client = ClinicalTrialsMCPClient(mcp_url=MCP_URL)
+    anchors = SearchAnchors.model_validate(json.loads(search_anchors))
+    raw_studies = await client.search(anchors)
+
+    trials = [
+        ClinicalTrialsMCPClient.normalize_trial(s).model_dump()
+        for s in raw_studies
+    ]
+
+    return ToolResult(
+        data={"trials": trials, "count": len(trials)},
+        metadata={"source": "clinicaltrials_mcp"},
+    )
+
+
+@tool
+async def refine_search_query(
+    context: ToolContext,
+    search_anchors: str,
+    result_count: str,
+) -> ToolResult:
+    """Refine search parameters when too many results returned.
+
+    Args:
+        context: Parlant tool context.
+        search_anchors: JSON string of current SearchAnchors.
+        result_count: Number of results from last search.
+    """
+    from trialpath.models.search_anchors import SearchAnchors
+    from trialpath.services.gemini_planner import GeminiPlanner
+
+    planner = GeminiPlanner(model=GEMINI_MODEL, api_key=GEMINI_API_KEY)
+    anchors = SearchAnchors.model_validate(json.loads(search_anchors))
+    refined = await planner.refine_search(anchors, int(result_count))
+
+    return ToolResult(
+        data=refined.model_dump(),
+        metadata={"action": "refine", "prev_count": int(result_count)},
+    )
+
+
+@tool
+async def relax_search_query(
+    context: ToolContext,
+    search_anchors: str,
+    result_count: str,
+) -> ToolResult:
+    """Relax search parameters when too few results returned.
+
+    Args:
+        context: Parlant tool context.
+        search_anchors: JSON string of current SearchAnchors.
+        result_count: Number of results from last search.
+    """
+    from trialpath.models.search_anchors import SearchAnchors
+    from trialpath.services.gemini_planner import GeminiPlanner
+
+    planner = GeminiPlanner(model=GEMINI_MODEL, api_key=GEMINI_API_KEY)
+    anchors = SearchAnchors.model_validate(json.loads(search_anchors))
+    relaxed = await planner.relax_search(anchors, int(result_count))
+
+    return ToolResult(
+        data=relaxed.model_dump(),
+        metadata={"action": "relax", "prev_count": int(result_count)},
+    )
+
+
+@tool
+async def evaluate_trial_eligibility(
+    context: ToolContext,
+    patient_profile: str,
+    trial_candidate: str,
+) -> ToolResult:
+    """Evaluate patient eligibility for a clinical trial using dual-model approach.
+
+    Medical criteria evaluated by MedGemma, structural by Gemini.
+
+    Args:
+        context: Parlant tool context.
+        patient_profile: JSON string of PatientProfile data.
+        trial_candidate: JSON string of TrialCandidate data.
+    """
+    from trialpath.services.gemini_planner import GeminiPlanner
+    from trialpath.services.medgemma_extractor import MedGemmaExtractor
+
+    profile = json.loads(patient_profile)
+    trial = json.loads(trial_candidate)
+
+    planner = GeminiPlanner(model=GEMINI_MODEL, api_key=GEMINI_API_KEY)
+    extractor = MedGemmaExtractor(
+        endpoint_url=MEDGEMMA_ENDPOINT_URL,
+        hf_token=HF_TOKEN,
+    )
+
+    # Step 1: Slice criteria into atomic items
+    criteria = await planner.slice_criteria(trial)
+
+    # Step 2: Evaluate each criterion with appropriate model
+    assessments = []
+    for criterion in criteria:
+        if criterion.get("category") == "medical":
+            result = await extractor.evaluate_medical_criterion(
+                criterion["text"], profile, []
+            )
+        else:
+            result = await planner.evaluate_structural_criterion(
+                criterion["text"], profile
+            )
+        assessments.append({**criterion, **result})
+
+    # Step 3: Aggregate into overall assessment
+    ledger = await planner.aggregate_assessments(profile, trial, assessments)
+
+    return ToolResult(
+        data=ledger.model_dump(),
+        metadata={"source": "dual_model", "criteria_count": len(criteria)},
+    )
+
+
+@tool
+async def analyze_gaps(
+    context: ToolContext,
+    patient_profile: str,
+    eligibility_ledgers: str,
+) -> ToolResult:
+    """Analyze eligibility gaps across all evaluated trials.
+
+    Args:
+        context: Parlant tool context.
+        patient_profile: JSON string of PatientProfile data.
+        eligibility_ledgers: JSON list of EligibilityLedger data.
+    """
+    from trialpath.services.gemini_planner import GeminiPlanner
+
+    planner = GeminiPlanner(model=GEMINI_MODEL, api_key=GEMINI_API_KEY)
+    profile = json.loads(patient_profile)
+    ledgers = json.loads(eligibility_ledgers)
+    gaps = await planner.analyze_gaps(profile, ledgers)
+
+    return ToolResult(
+        data={"gaps": gaps, "count": len(gaps)},
+        metadata={"source": "gemini"},
+    )
+
+
+ALL_TOOLS = [
+    extract_patient_profile,
+    generate_search_anchors,
+    search_clinical_trials,
+    refine_search_query,
+    relax_search_query,
+    evaluate_trial_eligibility,
+    analyze_gaps,
+]

trialpath/tests/test_tools.py

@@ -0,0 +1,276 @@
+"""TDD tests for Parlant tool functions."""
+import json
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from trialpath.agent.tools import (
+    ALL_TOOLS,
+    analyze_gaps,
+    evaluate_trial_eligibility,
+    extract_patient_profile,
+    generate_search_anchors,
+    refine_search_query,
+    relax_search_query,
+    search_clinical_trials,
+)
+
+
+@pytest.fixture
+def mock_context():
+    return MagicMock()
+
+
+class TestExtractPatientProfile:
+    """Test extract_patient_profile tool."""
+
+    @pytest.mark.asyncio
+    async def test_calls_medgemma_extractor(self, mock_context):
+        """Should call MedGemmaExtractor.extract with correct args."""
+        profile = {"patient_id": "P001", "diagnosis": {"primary_condition": "NSCLC"}}
+
+        with patch(
+            "trialpath.services.medgemma_extractor.MedGemmaExtractor"
+        ) as MockExtractor:
+            MockExtractor.return_value.extract = AsyncMock(return_value=profile)
+
+            result = await extract_patient_profile.function(
+                mock_context,
+                document_urls=json.dumps(["doc1.pdf"]),
+                metadata=json.dumps({"age": 52}),
+            )
+
+            MockExtractor.return_value.extract.assert_called_once()
+            assert result.data["patient_id"] == "P001"
+
+    @pytest.mark.asyncio
+    async def test_returns_tool_result_with_metadata(self, mock_context):
+        """ToolResult should contain source metadata."""
+        with patch(
+            "trialpath.services.medgemma_extractor.MedGemmaExtractor"
+        ) as MockExtractor:
+            MockExtractor.return_value.extract = AsyncMock(return_value={})
+
+            result = await extract_patient_profile.function(
+                mock_context,
+                document_urls=json.dumps(["a.pdf", "b.pdf"]),
+                metadata=json.dumps({}),
+            )
+
+            assert result.metadata["source"] == "medgemma"
+            assert result.metadata["doc_count"] == 2
+
+
+class TestGenerateSearchAnchors:
+    """Test generate_search_anchors tool."""
+
+    @pytest.mark.asyncio
+    async def test_calls_gemini_planner(self, mock_context):
+        """Should call GeminiPlanner.generate_search_anchors."""
+        from trialpath.models.search_anchors import SearchAnchors
+
+        mock_anchors = SearchAnchors(condition="NSCLC")
+
+        with patch(
+            "trialpath.services.gemini_planner.GeminiPlanner"
+        ) as MockPlanner:
+            MockPlanner.return_value.generate_search_anchors = AsyncMock(
+                return_value=mock_anchors
+            )
+
+            result = await generate_search_anchors.function(
+                mock_context,
+                patient_profile=json.dumps({"patient_id": "P001"}),
+            )
+
+            assert result.data["condition"] == "NSCLC"
+
+
+class TestSearchClinicalTrials:
+    """Test search_clinical_trials tool."""
+
+    @pytest.mark.asyncio
+    async def test_calls_mcp_client_and_normalizes(self, mock_context):
+        """Should call MCP client and normalize results."""
+        raw_study = {"nctId": "NCT001", "title": "Test Trial"}
+
+        with patch(
+            "trialpath.services.mcp_client.ClinicalTrialsMCPClient"
+        ) as MockClient:
+            MockClient.return_value.search = AsyncMock(return_value=[raw_study])
+            mock_trial = MagicMock()
+            mock_trial.model_dump.return_value = {
+                "nct_id": "NCT001", "title": "Test Trial"
+            }
+            MockClient.normalize_trial = MagicMock(return_value=mock_trial)
+
+            result = await search_clinical_trials.function(
+                mock_context,
+                search_anchors=json.dumps({"condition": "NSCLC"}),
+            )
+
+            assert result.data["count"] == 1
+            assert result.metadata["source"] == "clinicaltrials_mcp"
+
+
+class TestRefineSearchQuery:
+    """Test refine_search_query tool."""
+
+    @pytest.mark.asyncio
+    async def test_calls_gemini_refine(self, mock_context):
+        """Should call GeminiPlanner.refine_search."""
+        from trialpath.models.search_anchors import SearchAnchors
+
+        mock_refined = SearchAnchors(condition="NSCLC", biomarkers=["EGFR"])
+
+        with patch(
+            "trialpath.services.gemini_planner.GeminiPlanner"
+        ) as MockPlanner:
+            MockPlanner.return_value.refine_search = AsyncMock(
+                return_value=mock_refined
+            )
+
+            result = await refine_search_query.function(
+                mock_context,
+                search_anchors=json.dumps({"condition": "NSCLC"}),
+                result_count="100",
+            )
+
+            assert result.metadata["action"] == "refine"
+            assert result.metadata["prev_count"] == 100
+
+
+class TestRelaxSearchQuery:
+    """Test relax_search_query tool."""
+
+    @pytest.mark.asyncio
+    async def test_calls_gemini_relax(self, mock_context):
+        """Should call GeminiPlanner.relax_search."""
+        from trialpath.models.search_anchors import SearchAnchors
+
+        mock_relaxed = SearchAnchors(condition="NSCLC")
+
+        with patch(
+            "trialpath.services.gemini_planner.GeminiPlanner"
+        ) as MockPlanner:
+            MockPlanner.return_value.relax_search = AsyncMock(
+                return_value=mock_relaxed
+            )
+
+            result = await relax_search_query.function(
+                mock_context,
+                search_anchors=json.dumps({"condition": "NSCLC"}),
+                result_count="0",
+            )
+
+            assert result.metadata["action"] == "relax"
+
+
+class TestEvaluateTrialEligibility:
+    """Test evaluate_trial_eligibility tool."""
+
+    @pytest.mark.asyncio
+    async def test_dual_model_evaluation(self, mock_context):
+        """Should use MedGemma for medical and Gemini for structural criteria."""
+        from trialpath.models.eligibility_ledger import (
+            EligibilityLedger,
+            OverallAssessment,
+        )
+
+        mock_ledger = EligibilityLedger(
+            patient_id="P001",
+            nct_id="NCT001",
+            overall_assessment=OverallAssessment.LIKELY_ELIGIBLE,
+            criteria=[],
+            gaps=[],
+        )
+
+        with (
+            patch(
+                "trialpath.services.gemini_planner.GeminiPlanner"
+            ) as MockPlanner,
+            patch(
+                "trialpath.services.medgemma_extractor.MedGemmaExtractor"
+            ) as MockExtractor,
+        ):
+            MockPlanner.return_value.slice_criteria = AsyncMock(
+                return_value=[
+                    {
+                        "criterion_id": "inc_1",
+                        "type": "inclusion",
+                        "text": "EGFR mutation",
+                        "category": "medical",
+                    },
+                    {
+                        "criterion_id": "inc_2",
+                        "type": "inclusion",
+                        "text": "Age >= 18",
+                        "category": "structural",
+                    },
+                ]
+            )
+            MockExtractor.return_value.evaluate_medical_criterion = AsyncMock(
+                return_value={"decision": "met", "reasoning": "OK", "confidence": 0.9}
+            )
+            MockPlanner.return_value.evaluate_structural_criterion = AsyncMock(
+                return_value={"decision": "met", "reasoning": "OK", "confidence": 0.99}
+            )
+            MockPlanner.return_value.aggregate_assessments = AsyncMock(
+                return_value=mock_ledger
+            )
+
+            result = await evaluate_trial_eligibility.function(
+                mock_context,
+                patient_profile=json.dumps({"patient_id": "P001"}),
+                trial_candidate=json.dumps({"nct_id": "NCT001"}),
+            )
+
+            assert result.data["overall_assessment"] == "likely_eligible"
+            assert result.metadata["criteria_count"] == 2
+            MockExtractor.return_value.evaluate_medical_criterion.assert_called_once()
+            MockPlanner.return_value.evaluate_structural_criterion.assert_called_once()
+
+
+class TestAnalyzeGaps:
+    """Test analyze_gaps tool."""
+
+    @pytest.mark.asyncio
+    async def test_calls_gemini_gap_analysis(self, mock_context):
+        """Should call GeminiPlanner.analyze_gaps."""
+        mock_gaps = [
+            {
+                "description": "Brain MRI needed",
+                "recommended_action": "Upload MRI",
+                "clinical_importance": "high",
+                "affected_trial_count": 2,
+            }
+        ]
+
+        with patch(
+            "trialpath.services.gemini_planner.GeminiPlanner"
+        ) as MockPlanner:
+            MockPlanner.return_value.analyze_gaps = AsyncMock(return_value=mock_gaps)
+
+            result = await analyze_gaps.function(
+                mock_context,
+                patient_profile=json.dumps({}),
+                eligibility_ledgers=json.dumps([]),
+            )
+
+            assert result.data["count"] == 1
+            assert result.data["gaps"][0]["clinical_importance"] == "high"
+
+
+class TestAllToolsExported:
+    """Test ALL_TOOLS list completeness."""
+
+    def test_all_tools_has_7_entries(self):
+        """ALL_TOOLS should contain exactly 7 tools."""
+        assert len(ALL_TOOLS) == 7
+
+    def test_all_tools_are_tool_entries(self):
+        """Each item in ALL_TOOLS should be a ToolEntry."""
+        from parlant.sdk import ToolEntry
+
+        for t in ALL_TOOLS:
+            assert isinstance(t, ToolEntry), f"{t} is not a ToolEntry"