d77fe12cfc
Source layer (5 court sources), processing pipeline (parse/extract/chunk/embed/graph), property graph with 8 node types, juror subgraph queries with 6 personas, orchestrator with bootstrap/watch/backfill/audit/process modes, 170 tests. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
"""AusLaw MCP source — targeted retrieval via external MCP server.
|
|
|
|
This is NOT a direct HTTP source. It requires auslaw-mcp to be running
|
|
as a separate process. Use mcp-python-sdk to connect.
|
|
|
|
Repo: https://github.com/russellbrenner/auslaw-mcp
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from aucourt_ingest.models import FetchQueueItem, FetchStatus, RawDocument
|
|
from aucourt_ingest.sources.base import BaseSource
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AusLawMCPSource(BaseSource):
|
|
"""AusLaw MCP — targeted case retrieval.
|
|
|
|
NOT implemented yet — requires:
|
|
1. auslaw-mcp server running (npm install -g auslaw-mcp or docker)
|
|
2. mcp-python-sdk client connection
|
|
3. Tool call wrappers for search_austlii, fetch_document_text, etc.
|
|
|
|
This source is used for gap-filling, not bulk ingestion.
|
|
"""
|
|
|
|
source_id = "auslaw_mcp"
|
|
|
|
async def discover(self, page: int = 1, **kwargs) -> list[FetchQueueItem]:
|
|
raise NotImplementedError("AusLaw MCP is for targeted retrieval, not bulk discovery")
|
|
|
|
async def fetch(self, url: str, **kwargs) -> RawDocument:
|
|
raise NotImplementedError(
|
|
"AusLaw MCP requires external server. "
|
|
"Install auslaw-mcp and implement MCP client in Stage 3b."
|
|
)
|
|
|
|
async def fetch_by_citation(self, citation: str) -> RawDocument | None:
|
|
"""Fetch a specific case by its medium neutral citation.
|
|
|
|
TODO: Implement via MCP tool call:
|
|
result = await mcp.call("fetch_document_text", {"url": austlii_url})
|
|
"""
|
|
logger.warning(f"AusLaw MCP fetch_by_citation not yet implemented: {citation}")
|
|
return None
|
|
|
|
async def search(self, query: str, jurisdiction: str | None = None,
|
|
limit: int = 20) -> list[FetchQueueItem]:
|
|
"""Search for cases matching a query.
|
|
|
|
TODO: Implement via MCP tool call:
|
|
results = await mcp.call("search_austlii", {"query": query, ...})
|
|
"""
|
|
logger.warning(f"AusLaw MCP search not yet implemented: {query}")
|
|
return []
|