"""Tool registry for the Cornell LII legal-research agent. A second Agent Mode toolset (see :mod:`webapp.agent_loop`), parallel to :mod:`webapp.agent_tools` but pointed at **US federal/state law** instead of the uploaded packet. Each tool is a thin wrapper over the :mod:`cornell_lii` library — resolving and linking CFR/USC citations to the pages Cornell Legal Information Institute hosts, verifying them and full-text-searching the public **eCFR API**. Design notes: * **Packet-independent.** These tools don't touch the uploaded PDF; the loop passes a light :class:`LiiContext` (carrying only a per-turn network cache) straight through. * **Links + excerpts, not full text.** LII content isn't scraped (its ``robots.txt`` disallows it). The agent returns citations, official LII links, and short eCFR search excerpts — it must not invent statutory text. The system prompt enforces this. * **Live external calls.** ``search_regulations`` and citation verification hit the eCFR API / ``law.cornell.edu`` over the network (httpx, 30s timeout). Results are memoized on the turn's :class:`LiiContext` so retries within a turn don't re-fetch. * **Same contract as agent_tools.** Every tool returns a JSON-serializable dict; failures come back as ``{"error": ...}`` so the model can recover. Outputs are bounded. """ from __future__ import annotations import re from dataclasses import dataclass, field from typing import Callable from webapp.agent_loop import Toolkit from webapp.agent_tools import Tool, _clip, observation_text # Output caps — keep any single observation small enough to feed back into an 8K ctx. _MAX_RESULTS = 8 # eCFR search hits returned to the model _EXCERPT_CHARS = 400 # per-hit excerpt slice @dataclass class LiiContext: """Context for the LII tools. Unlike the packet agent there is no ``upload_id``; the only state is an optional per-turn cache so repeated network lookups are free.""" _cache: dict = field(default_factory=dict) # --------------------------------------------------------------------------- # # Small helpers # --------------------------------------------------------------------------- # def _cached(ctx: LiiContext, key: tuple, fn: Callable[[], object]) -> object: """Memoize a (possibly networked) lookup on the turn's context cache.""" cache = ctx._cache if key not in cache: cache[key] = fn() return cache[key] def _need(value, name: str) -> str: """Coerce a required scalar arg to a non-empty string or raise ``ValueError``.""" s = str(value).strip() if value is not None else "" if not s: raise ValueError(f"Missing required argument '{name}'.") return s # --------------------------------------------------------------------------- # # Tools # --------------------------------------------------------------------------- # def _search_regulations(ctx: LiiContext, query: str = "", limit: int = _MAX_RESULTS) -> dict: from cornell_lii.client import HttpError from cornell_lii.ecfr import search_federal q = _need(query, "query") try: n = min(max(int(limit), 1), _MAX_RESULTS) except (TypeError, ValueError): n = _MAX_RESULTS try: hits = _cached(ctx, ("search", q, n), lambda: search_federal(q, limit=n)) except HttpError as e: return {"error": f"eCFR search failed: {e}"} results = [ { "citation": h.get("citation"), "label": h.get("label"), "url": h.get("url"), "excerpt": _clip(h.get("excerpt") or "", _EXCERPT_CHARS), } for h in hits ] return {"query": q, "count": len(results), "results": results} def _resolve_cfr( ctx: LiiContext, citation: str | None = None, title: str | None = None, section: str | None = None, verify: bool = True, ) -> dict: from cornell_lii.citations import CitationError, parse_cfr from cornell_lii.client import HttpError from cornell_lii.ecfr import verify_cfr from cornell_lii.urls import cfr_section_url try: if citation: cite = parse_cfr(citation) title, section = cite.title, cite.section else: title, section = _need(title, "title"), _need(section, "section") except (CitationError, ValueError) as e: return {"error": str(e)} out: dict = { "citation": f"{title} CFR {section}", "title": title, "section": section, "url": cfr_section_url(title, section), } if verify: try: v = _cached(ctx, ("verify_cfr", title, section), lambda: verify_cfr(title, section)) out.update(verified=v["valid"], heading=v["heading"], date=v["date"]) if not v["valid"] and v.get("detail"): out["detail"] = v["detail"] except HttpError as e: out["verified"] = None out["detail"] = f"eCFR verify failed: {e}" return out def _resolve_usc( ctx: LiiContext, citation: str | None = None, title: str | None = None, section: str | None = None, verify: bool = True, ) -> dict: from cornell_lii.citations import CitationError, parse_usc from cornell_lii.client import url_exists from cornell_lii.urls import usc_section_url try: if citation: cite = parse_usc(citation) title, section = cite.title, cite.section else: title, section = _need(title, "title"), _need(section, "section") except (CitationError, ValueError) as e: return {"error": str(e)} url = usc_section_url(title, section) out: dict = { "citation": f"{title} USC {section}", "title": title, "section": section, "url": url, } if verify: out["verified"] = _cached(ctx, ("usc_exists", url), lambda: url_exists(url)) return out def _browse_cfr(ctx: LiiContext, title: str = "", part: str | None = None) -> dict: from cornell_lii.urls import cfr_part_url, cfr_title_url try: t = _need(title, "title") except ValueError as e: return {"error": str(e)} if part: p = str(part).strip() return {"citation": f"{t} CFR Part {p}", "url": cfr_part_url(t, p)} return {"citation": f"{t} CFR", "url": cfr_title_url(t)} def _state_materials(ctx: LiiContext, state: str = "") -> dict: from cornell_lii.states import resolve_state from cornell_lii.urls import state_portal_url, state_regs_url try: slug = resolve_state(_need(state, "state")) except ValueError as e: return {"error": str(e)} return { "state": slug.replace("_", " ").title(), "slug": slug, "portal_url": state_portal_url(slug), "regs_url": state_regs_url(slug), } def _mcl_find(ctx: LiiContext, topic: str = "") -> dict: from cornell_lii import michigan as mi try: q = _need(topic, "topic") except ValueError as e: return {"error": str(e)} results = mi.find_starting_points(q, limit=6) return { "query": q, "count": len(results), "results": results, "search_url": mi.mcl_search_url(), "note": "These are the major acts/chapters for this topic. For the specific section " "whose text matches a keyword, use mcl_search; then read it with mcl_text.", } def _mcl_search( ctx: LiiContext, query: str = "", chapter: str = "", popular_name: str = "", ) -> dict: from cornell_lii import mileg from cornell_lii.client import HttpError q = (query or "").strip() if not (q or chapter or popular_name): return {"error": "Provide 'query' keywords to full-text search the MCL (optionally " "narrow with 'chapter' or 'popular_name')."} try: res = _cached( ctx, ("mcl_search", q, chapter, popular_name), lambda: mileg.search_mcl(q, chapter=chapter, popular_name=popular_name, limit=12), ) except HttpError as e: return {"error": f"MCL search failed: {e}. Try different keywords."} out = dict(res) out["note"] = ( "Full-text matches across the whole MCL. Read a promising section's text with " "mcl_text(section=...) — do not re-run this search with the same query." ) return out def _mcl_lookup(ctx: LiiContext, citation: str | None = None, section: str | None = None) -> dict: from cornell_lii import michigan as mi sec = mi.parse_mcl(str(section)) if section else (mi.parse_mcl(str(citation)) if citation else None) if not sec: return {"error": "Provide an MCL section like '691.1407' (as 'section', or inside " "'citation'). For a topic instead of a number, use mcl_find."} out: dict = { "citation": f"MCL {sec}", "section": sec, "url": mi.mcl_section_url(sec), "source": "legislature.mi.gov", } _attach_mcl_context(out, sec) return out def _attach_mcl_context(out: dict, sec: str) -> None: """Add the subject chapter and containing act for an MCL section to a result dict.""" from cornell_lii import michigan as mi ch = mi.chapter_for_section(sec) if ch: out["chapter"] = {"prefix": ch["prefix"], "name": ch["name"], "url": mi.mcl_chapter_url(ch["prefix"])} act = mi.act_for_section(sec) if act: n, y = act["act"] out["act"] = {"name": act["name"], "range": f"{act['start']}–{act['end']}", "citation": f"Act {n} of {y}", "url": mi.mcl_act_url(n, y)} # One page of MCL section text per read (chars), matching the agenda agent's get_item_text # convention so a long section is read across calls instead of re-read identically; and the # (effectively unbounded) cap we fetch the full section at, so paging windows a cached body. _MCL_WINDOW = 2600 _MCL_FETCH_CAP = 200_000 def _mcl_text( ctx: LiiContext, section: str | None = None, citation: str | None = None, offset: int = 0, ) -> dict: from cornell_lii import mileg from cornell_lii import michigan as mi from cornell_lii.client import HttpError sec = mi.parse_mcl(str(section)) if section else (mi.parse_mcl(str(citation)) if citation else None) if not sec: return {"error": "Provide an MCL section like '257.625' (as 'section', or inside " "'citation'). To find a section by topic, use mcl_find / mcl_outline first."} try: # Fetch the FULL section once (cached per section), then page through it locally — # a long section reads across calls via `offset` rather than being re-fetched and # re-truncated (which previously lured the model into re-reading the same text). res = _cached( ctx, ("mcl_text", sec), lambda: mileg.get_section_text(sec, char_limit=_MCL_FETCH_CAP) ) except HttpError as e: return {"error": f"Could not read MCL {sec}: {e}. Check the section number."} if not res.get("found"): return {"error": f"No MCL section {sec} found. Use mcl_outline to list valid sections."} full = res.get("text", "") total = len(full) try: start = max(0, int(offset)) except (TypeError, ValueError): start = 0 window = full[start : start + _MCL_WINDOW] next_offset = start + len(window) has_more = next_offset < total out = { "citation": f"MCL {sec}", "section": sec, "heading": res.get("heading", ""), "text": window + (" …[more — call mcl_text again with offset=next_offset]" if has_more else ""), "offset": start, "next_offset": next_offset, "has_more": has_more, "total_chars": total, "url": res.get("url"), "source": "legislature.mi.gov", } if has_more: out["note"] = ( f"This section is long: showing characters {start}–{next_offset} of {total}. To " f"read the next part, call mcl_text with section='{sec}' and offset={next_offset}. " "Do not re-read the same offset." ) _attach_mcl_context(out, sec) return out def _mcl_outline(ctx: LiiContext, act: str | None = None, chapter: str | None = None) -> dict: from cornell_lii import mileg from cornell_lii.client import HttpError if chapter: object_name = f"mcl-chap{str(chapter).strip()}" label = f"MCL chapter {str(chapter).strip()}" elif act: nums = re.findall(r"\d+", str(act)) year = next((n for n in nums if len(n) == 4), None) number = next((n for n in nums if n != year), None) if not (number and year): return {"error": "Give the act as 'Act 267 of 1976' (or '267 of 1976'), or pass " "a 'chapter' like '257'."} object_name = f"mcl-Act-{number}-of-{year}" label = f"Act {number} of {year}" else: return {"error": "Provide an 'act' (e.g. 'Act 267 of 1976') or a 'chapter' (e.g. '257')."} try: res = _cached(ctx, ("mcl_outline", object_name), lambda: mileg.list_sections(object_name)) except HttpError as e: return {"error": f"Could not list {label}: {e}."} if not res.get("count"): return {"error": f"No sections found for {label}. Check the act/chapter."} return { "target": label, "count": res["count"], "sections": res["sections"], "url": res["url"], "note": "Use mcl_text on a section number to read its full text.", } # --------------------------------------------------------------------------- # # Registry, catalog, dispatch # --------------------------------------------------------------------------- # TOOLS: list[Tool] = [ Tool( name="search_regulations", description="Full-text search the federal Code of Federal Regulations (eCFR) and " "get matching sections as Cornell LII links with short excerpts. Use this first " "for topic questions when you don't already have a citation.", parameters={ "type": "object", "properties": { "query": {"type": "string", "description": "Search terms, e.g. 'sexual harassment'."}, "limit": {"type": "integer", "description": f"Max hits (1-{_MAX_RESULTS}). Default {_MAX_RESULTS}."}, }, "required": ["query"], }, run=_search_regulations, ), Tool( name="resolve_cfr", description="Resolve a federal regulation (CFR) citation to its Cornell LII page " "and (by default) verify it exists via eCFR, returning the section heading. Pass " "either a full 'citation' (e.g. '29 CFR 1604.11') or 'title' + 'section'.", parameters={ "type": "object", "properties": { "citation": {"type": "string", "description": "Full citation, e.g. '29 CFR 1604.11'."}, "title": {"type": "string", "description": "CFR title number, e.g. '29'."}, "section": {"type": "string", "description": "Section, e.g. '1604.11'."}, "verify": {"type": "boolean", "description": "Confirm via eCFR (default true)."}, }, }, run=_resolve_cfr, ), Tool( name="resolve_usc", description="Resolve a US Code (federal statute) citation to its Cornell LII page " "and (by default) verify the page resolves. Pass either a full 'citation' (e.g. " "'42 USC 1983') or 'title' + 'section'.", parameters={ "type": "object", "properties": { "citation": {"type": "string", "description": "Full citation, e.g. '42 USC 1983'."}, "title": {"type": "string", "description": "USC title number, e.g. '42'."}, "section": {"type": "string", "description": "Section, e.g. '1983'."}, "verify": {"type": "boolean", "description": "Confirm the LII page exists (default true)."}, }, }, run=_resolve_usc, ), Tool( name="browse_cfr", description="Get the Cornell LII browse link for a whole CFR title, or a part " "within it. Use when the user wants the index/part rather than a single section.", parameters={ "type": "object", "properties": { "title": {"type": "string", "description": "CFR title number, e.g. '29'."}, "part": {"type": "string", "description": "Optional part number, e.g. '1604'."}, }, "required": ["title"], }, run=_browse_cfr, ), Tool( name="state_materials", description="Get Cornell LII links for a US state's legal materials: the state " "portal page and its LII-hosted administrative code. Accepts a state name or USPS " "abbreviation (e.g. 'Maine' or 'ME').", parameters={ "type": "object", "properties": { "state": {"type": "string", "description": "State name or USPS abbreviation."}, }, "required": ["state"], }, run=_state_materials, ), Tool( name="mcl_find", description="MICHIGAN statutes (Michigan Compiled Laws / MCL): given a topic or an " "act name, return the right code(s)/act(s) to start in, with official " "legislature.mi.gov links and section ranges. Use this FIRST for any Michigan-law " "question when you don't already have an MCL number.", parameters={ "type": "object", "properties": { "topic": {"type": "string", "description": "A topic or act name, e.g. 'drunk " "driving', 'open meetings', 'Elliott-Larsen'."}, }, "required": ["topic"], }, run=_mcl_find, ), Tool( name="mcl_search", description="MICHIGAN statutes: FULL-TEXT search the Michigan Compiled Laws on " "legislature.mi.gov by keyword — finds the specific section(s) whose text matches, " "across the ENTIRE MCL (not just the major acts mcl_find knows). Returns section " "numbers + catchlines; then read a section with mcl_text. Use this when mcl_find's " "starting points are too broad or miss the topic. Optionally narrow with 'chapter' " "(e.g. '15') or 'popular_name' (e.g. 'Freedom of Information Act').", parameters={ "type": "object", "properties": { "query": {"type": "string", "description": "Keywords to full-text search the " "MCL, e.g. 'public notice meeting'."}, "chapter": {"type": "string", "description": "Restrict to an MCL chapter " "prefix, e.g. '15'."}, "popular_name": {"type": "string", "description": "Restrict to an act's popular " "name, e.g. 'Open Meetings Act'."}, }, "required": ["query"], }, run=_mcl_search, ), Tool( name="mcl_lookup", description="MICHIGAN statutes: resolve an MCL section to its page on " "legislature.mi.gov and identify its subject chapter and the specific act it " "belongs to (link + structure only, no text). Pass an MCL section ('691.1407') or " "a citation ('MCL 691.1407'). To READ the text, use mcl_text.", parameters={ "type": "object", "properties": { "section": {"type": "string", "description": "MCL section number, e.g. '691.1407'."}, "citation": {"type": "string", "description": "Full MCL citation, e.g. 'MCL 691.1407'."}, }, }, run=_mcl_lookup, ), Tool( name="mcl_text", description="MICHIGAN statutes: read the ACTUAL TEXT of an MCL section from " "legislature.mi.gov — the heading and body (with amendment history). Use this to " "quote or answer substantive questions about what a Michigan law says. Pass an MCL " "section ('257.625') or citation ('MCL 257.625'). Long sections return one page of " "text at a time; if has_more is true, call again with offset=next_offset to read the " "rest — never re-read the same offset.", parameters={ "type": "object", "properties": { "section": {"type": "string", "description": "MCL section number, e.g. '257.625'."}, "citation": {"type": "string", "description": "Full MCL citation, e.g. 'MCL 257.625'."}, "offset": { "type": "integer", "description": "Char offset to resume reading a long section (use next_offset from the prior call; default 0).", }, }, }, run=_mcl_text, ), Tool( name="mcl_outline", description="MICHIGAN statutes: list the sections (number + heading) of a Michigan " "act or chapter, so you can find the right section to read. Pass an 'act' (e.g. " "'Act 267 of 1976') or a 'chapter' (e.g. '257'). Then call mcl_text on a section.", parameters={ "type": "object", "properties": { "act": {"type": "string", "description": "Act citation, e.g. 'Act 267 of 1976'."}, "chapter": {"type": "string", "description": "MCL chapter prefix, e.g. '257'."}, }, }, run=_mcl_outline, ), ] _BY_NAME = {t.name: t for t in TOOLS} # final_answer is special-cased by the loop (it ends the turn), but advertised in the # catalog so the model sees it alongside the real tools. FINAL_ANSWER = "final_answer" def tool_catalog() -> list[dict]: """OpenAI-style ``{name, description, parameters}`` schema for every LII tool, plus the terminal ``final_answer`` — what the agent loop shows the model.""" cat = [ {"name": t.name, "description": t.description, "parameters": t.parameters} for t in TOOLS ] cat.append( { "name": FINAL_ANSWER, "description": "Give the user your final answer and end the turn. Use this as " "soon as you have enough to answer.", "parameters": { "type": "object", "properties": { "answer": {"type": "string", "description": "The answer, in markdown."} }, "required": ["answer"], }, } ) return cat def run_tool(name: str, args: dict, ctx: LiiContext) -> dict: """Dispatch one LII tool call. Returns a JSON-serializable result dict (always — a failed/unknown call comes back as ``{"error": ...}`` so the model can recover).""" tool = _BY_NAME.get(name) if tool is None: return {"error": f"Unknown tool '{name}'. Available: {', '.join(_BY_NAME)}."} if not isinstance(args, dict): return {"error": "Tool args must be a JSON object."} try: return tool.run(ctx, **args) except TypeError as e: return {"error": f"Bad arguments for '{name}': {e}"} except Exception as e: # noqa: BLE001 return {"error": f"Tool '{name}' failed: {type(e).__name__}: {e}"} def result_summary(tool: str, result: dict) -> str: """A one-line, human-readable gist of a tool result for the live UI (raw JSON stays available behind a disclosure). Best-effort — empty string when nothing fits.""" if not isinstance(result, dict): return "" if result.get("error"): return f"error: {str(result['error'])[:80]}" if tool == "mcl_text" and result.get("section"): return f"read MCL {result['section']}" + (" (more)" if result.get("has_more") else "") if tool == "mcl_outline" and isinstance(result.get("count"), int): n = result["count"] return f"{n} section{'' if n == 1 else 's'}" if tool == "mcl_search" and isinstance(result.get("count"), int): n = result["count"] return f"{n} MCL match{'' if n == 1 else 'es'}" if isinstance(result.get("count"), int): n = result["count"] return f"{n} result{'' if n == 1 else 's'}" if "verified" in result: cite = result.get("citation") or "citation" verified = result["verified"] state = "verified" if verified else ("unverified" if verified is False else "link ready") return f"{cite} · {state}" if result.get("url"): return result.get("citation") or "link ready" if result.get("portal_url"): return result.get("state") or "links ready" return "" def _system_prompt(ctx: LiiContext) -> str: import json from cornell_lii import michigan as mi catalog = json.dumps(tool_catalog(), ensure_ascii=False, indent=2) return ( "You are the Cornell LII legal-research agent. You help the user find and link US " "federal and state law — federal regulations (CFR), federal statutes (US Code), " "state legal materials, and Michigan statutes (MCL) — using the Cornell Legal " "Information Institute, the public eCFR API, and the Michigan Legislature.\n\n" "You work by calling tools, one at a time, and reasoning over what they return. For " "federal topic questions with no citation in hand, start with search_regulations; " "when you have a federal citation, use resolve_cfr / resolve_usc to link and verify " "it.\n\n" "Available tools (JSON Schema):\n" + catalog + "\n\n" "MICHIGAN LAW (MCL) — you have the structure baked in AND can read the text, so " "route confidently:\n" "- For a Michigan-law TOPIC or act name, call mcl_find to get the right code/act.\n" "- To find the SPECIFIC section whose text matches a keyword (across the whole MCL, " "not just the major acts), call mcl_search — best when mcl_find is too broad or the " "topic is narrow.\n" "- To see which sections an act/chapter contains, call mcl_outline (act or chapter) " "and pick the right section by its heading.\n" "- To READ a section's actual text (and quote it / answer what the law says), call " "mcl_text. Use mcl_lookup only when you just need the link + structure, not the text.\n" "- Typical flow for a substantive Michigan question: mcl_find or mcl_search → " "mcl_text (read it). legislature.mi.gov hosts the full MCL text (Cornell LII does not). Use the " "MCL map below to start, but DON'T invent section numbers — confirm via the tools.\n" + mi.cheatsheet() + "\n\n" "PROTOCOL — follow exactly:\n" "- Respond with a SINGLE JSON object and nothing else (no prose, no code fence).\n" '- Shape: {"thought": "", "tool": "", "args": {}}\n' "- Call exactly one tool per step. Read its result, then decide the next step.\n" f'- When you can answer, call "{FINAL_ANSWER}" with an "answer" in markdown.\n' "- If a tool returns an error, adjust your args or try another tool.\n\n" "CITING SOURCES — always link, using markdown:\n" "- Every citation MUST be a markdown link to the source URL from a tool result, e.g. " "[42 CFR 483.10](https://www.law.cornell.edu/cfr/text/42/483.10) or " "[MCL 15.263](https://www.legislature.mi.gov/Laws/MCL?objectName=mcl-15-263). Write " "the citation as the link text and the tool's URL as the target.\n" "- Use the exact URL the tool returned — never guess, shorten, or fabricate a link. " "If a tool gave no URL for a claim, don't link it (and say it's unverified).\n" "- When you quote or rely on several sources, you may add a short '## Sources' list " "of markdown links at the end in addition to the inline links.\n\n" "IMPORTANT — what you can and cannot do:\n" "- For MICHIGAN (MCL) you CAN read full section text via mcl_text — quote it and " "answer what the law says, always citing the section. For FEDERAL law you have only " "citations, official Cornell LII links, and SHORT eCFR search excerpts — NOT full " "text. Never invent or paraphrase legal text you have not seen in a tool result, and " "never invent citation numbers.\n" "- Every legal claim in your answer must point to a markdown link you obtained from a " "tool. Prefer verified citations; if a citation could not be verified, say so.\n" "- Close with a brief attribution: federal/state links via Cornell LII " "(law.cornell.edu, eCFR public domain); Michigan text via the Michigan Legislature " "(legislature.mi.gov)." ) # The toolkit the loop plugs in for the Cornell LII agent. TOOLKIT = Toolkit( system_prompt=_system_prompt, run_tool=run_tool, observation_text=observation_text, result_summary=result_summary, final_answer=FINAL_ANSWER, )