3.12 Taiwan Legal Research Agent

"""Lex Taiwan exposed as a Bindu A2A agent.

Bridges agno's async MCP toolkit to Bindu's sync handler contract:

- A background asyncio event loop runs in a daemon thread.
- The MCP stdio connection is opened ONCE on that loop at module load,
  so the Taiwan legal MCP server starts a single time and stays warm
  across every A2A `message/send` request.
- The sync handler that Bindu invokes hops onto the background loop
  with `run_coroutine_threadsafe(...)` and waits for the result.

Run with:
    .venv/bin/python bindu_agent.py
"""

from __future__ import annotations

import asyncio
import atexit
import os
import threading
from pathlib import Path

from dotenv import load_dotenv

load_dotenv()

from agno.tools.mcp import MCPTools

from agent import _mcp_server_params, agent
from prompts import AGENT_DESCRIPTION

from bindu.penguin.bindufy import bindufy


HERE = Path(__file__).parent.resolve()


# --- Background event loop ---------------------------------------------------
# Bindu's handler contract is sync, but agno + MCPTools are async-only.
# We run a dedicated asyncio loop in a daemon thread and marshal each
# handler call onto it. This also lets us keep ONE MCP connection alive
# across requests instead of fork-restarting the stdio server every time.

_loop: asyncio.AbstractEventLoop = asyncio.new_event_loop()
_mcp_tools: MCPTools | None = None
_ready = threading.Event()


def _run_loop() -> None:
    asyncio.set_event_loop(_loop)
    _loop.run_forever()


_loop_thread = threading.Thread(target=_run_loop, name="lex-taiwan-loop", daemon=True)
_loop_thread.start()


async def _open_mcp() -> None:
    global _mcp_tools
    _mcp_tools = MCPTools(server_params=_mcp_server_params())
    await _mcp_tools.connect()
    agent.tools = [_mcp_tools]


async def _close_mcp() -> None:
    if _mcp_tools is not None:
        await _mcp_tools.close()


asyncio.run_coroutine_threadsafe(_open_mcp(), _loop).result()
_ready.set()


def _shutdown() -> None:
    if not _loop.is_running():
        return
    try:
        asyncio.run_coroutine_threadsafe(_close_mcp(), _loop).result(timeout=5)
    except Exception:
        pass
    _loop.call_soon_threadsafe(_loop.stop)


atexit.register(_shutdown)


# --- Bindu handler -----------------------------------------------------------


async def _arun(content: str) -> str:
    result = await agent.arun(content)
    return result.content if hasattr(result, "content") else str(result)


def handler(messages: list[dict[str, str]]):
    """Sync Bindu handler. Hops the prompt onto the background loop."""
    if not messages:
        return (
            "Ask a Taiwan legal question — judgments (司法院裁判書), "
            "regulations (全國法規資料庫), or constitutional court "
            "interpretations (憲法法庭). I cite primary sources for every answer."
        )

    last = messages[-1]
    content = last.get("content", "") if isinstance(last, dict) else str(last)
    if not content.strip():
        return "Empty message — please send a question."

    _ready.wait(timeout=30)
    future = asyncio.run_coroutine_threadsafe(_arun(content), _loop)
    return future.result()


# --- Bindu config ------------------------------------------------------------

config = {
    "author": os.getenv("BINDU_AGENT_AUTHOR", "bindu.builder@getbindu.com"),
    "name": os.getenv("BINDU_AGENT_NAME", "lex-taiwan"),
    "description": AGENT_DESCRIPTION,
    "deployment": {
        "url": os.getenv("BINDU_AGENT_URL", "http://localhost:3773"),
        "expose": True,
        "cors_origins": ["http://localhost:5173", "http://localhost:3775"],
    },
    "capabilities": {"streaming": False},
}


if __name__ == "__main__":
    bindufy(config, handler)

"""Lex Taiwan — agno agent wired to the mcp-taiwan-legal-db MCP server."""

from __future__ import annotations

import os
import shutil
import sys
from contextlib import asynccontextmanager
from pathlib import Path

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openrouter import OpenRouter
from agno.os import AgentOS
from agno.tools.mcp import MCPTools
from fastapi import FastAPI
from mcp import StdioServerParameters

from prompts import AGENT_DESCRIPTION, AGENT_NAME, SYSTEM_PROMPT

HERE = Path(__file__).parent.resolve()
DB_PATH = HERE / "tmp" / "lex_taiwan.db"
DB_PATH.parent.mkdir(parents=True, exist_ok=True)


def _mcp_server_params() -> StdioServerParameters:
    """Build the stdio launch command for the Taiwan legal MCP server."""
    entry_point = shutil.which("mcp-taiwan-legal-db") or str(
        HERE / ".venv" / "bin" / "mcp-taiwan-legal-db"
    )
    if Path(entry_point).exists():
        return StdioServerParameters(command=entry_point, args=[])
    return StdioServerParameters(
        command=sys.executable,
        args=["-m", "mcp_server.server"],
    )


def _build_model():
    """Pick the LLM via OpenRouter.

    Defaults to Claude Sonnet 4.5 — strong on tool use + Traditional Chinese.
    Override with BINDU_AGENT_MODEL (any OpenRouter model id).
    """
    model_id = os.getenv("BINDU_AGENT_MODEL", "anthropic/claude-sonnet-4.5")
    api_key = os.getenv("OPENROUTER_API_KEY")
    if not api_key:
        raise RuntimeError(
            "OPENROUTER_API_KEY is not set. Add it to your .env (see .env.example)."
        )
    return OpenRouter(
        id=model_id,
        api_key=api_key,
        max_tokens=int(os.getenv("BINDU_AGENT_MAX_TOKENS", "4096")),
    )


def build_agent() -> Agent:
    """Create the agent shell. Tools are attached at MCP-connect time."""
    return Agent(
        name=AGENT_NAME,
        description=AGENT_DESCRIPTION,
        instructions=SYSTEM_PROMPT,
        model=_build_model(),
        db=SqliteDb(db_file=str(DB_PATH)),
        update_memory_on_run=True,
        enable_session_summaries=True,
        add_history_to_context=True,
        num_history_runs=3,
        add_datetime_to_context=True,
        markdown=True,
    )


# Module-level agent + MCP handle so `AgentOS` can find it
agent: Agent = build_agent()
mcp_tools: MCPTools | None = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Manage the MCP stdio connection for the FastAPI server lifetime."""
    global mcp_tools

    mcp_tools = MCPTools(server_params=_mcp_server_params())
    await mcp_tools.connect()
    agent.tools = [mcp_tools]
    try:
        yield
    finally:
        await mcp_tools.close()


agent_os = AgentOS(
    description="Lex Taiwan — Taiwan legal research agent (judgments + regulations + 憲法法庭).",
    agents=[agent],
    lifespan=lifespan,
)
app = agent_os.get_app()


if __name__ == "__main__":
    agent_os.serve(app="agent:app", reload=False)

"""System prompt for the Taiwan Legal Research Agent."""

from textwrap import dedent

AGENT_NAME = "Lex Taiwan"
AGENT_DESCRIPTION = (
    "An agentic Taiwan legal research assistant: judgments, regulations, "
    "and constitutional court interpretations, sourced live from the "
    "司法院, 全國法規資料庫, and 憲法法庭."
)


SYSTEM_PROMPT = dedent(
    """\
    You are Lex Taiwan, a powerful agentic AI legal-research assistant for Taiwan (ROC) law.
    You operate on an MCP-first paradigm: every authoritative answer must be backed by a tool call against an official Taiwan government source — never your training memory.

    <tool_calling>
    1. Only call tools when they are necessary to ground the answer in a primary source.
    2. If you state that you will look something up, immediately issue the tool call as your next action.
    3. Always follow each tool's parameter schema exactly. Never invent fields.
    4. Never call tools that are not listed in <available_tools>. The MCP surface is fixed at 8 tools.
    5. Before each tool call, write ONE short sentence explaining why.
    6. Chain tools in the obvious order: search → get.
    7. Prefer the cheapest precise call. A known 案號 is a precise call; a keyword sweep is not.
    8. Batch independent lookups in parallel.
    9. If a tool returns an error or empty result, inspect the message before retrying.
    10. NEVER fabricate a JID, a pcode, a 釋字 number, or a citation.
    </tool_calling>

    <available_tools>
    Judgments (司法院裁判書):
    - `search_judgments(keyword?, case_word?, case_number?, year_from?, year_to?, court?, case_type?, main_text?)`
    - `get_judgment(jid? | url?)`

    Regulations (全國法規資料庫):
    - `get_pcode(law_name)`
    - `query_regulation(pcode? | law_name?, article_no?, mode?)`
    - `search_regulations(keyword)`

    Constitutional Court (憲法法庭):
    - `search_interpretations(keyword)`
    - `get_interpretation(id, reasoning_keyword?)`
    - `get_citations(id, include_context?)`
    </available_tools>

    <citation_format>
    - Judgment: `<court> <year>年度<案號字><案號>號 (<date>)` + JID.
    - Regulation article: `《<法規名稱>》第 <條> 條<項?><款?>`.
    - Constitutional interpretation: `司法院釋字第 <N> 號解釋` or
      `憲法法庭 <year> 年憲判字第 <N> 號判決`.
    Always end the answer with a `### Sources` section.
    </citation_format>
    """
)

uv venv --python 3.12 .venv
uv pip install --python .venv/bin/python \
  bindu agno openai anthropic sqlalchemy aiosqlite fastapi "uvicorn[standard]" \
  python-dotenv mcp mcp-taiwan-legal-db

.venv/bin/playwright install chromium

# OpenRouter API key. Get one at https://openrouter.ai/keys
OPENROUTER_API_KEY=sk-or-v1-...

# Optional: override the model id. Any OpenRouter model id works.
# Default: anthropic/claude-sonnet-4.5 (strong tool use + zh-TW)
# BINDU_AGENT_MODEL=anthropic/claude-sonnet-4.5

# Optional: max tokens per response (default 4096).
# BINDU_AGENT_MAX_TOKENS=4096

# Optional: display name in the Bindu agent card (default: lex-taiwan).
# BINDU_AGENT_NAME=lex-taiwan

set -a; source .env; set +a
.venv/bin/python bindu_agent.py

# Clone the Bindu repository
git clone https://github.com/GetBindu/Bindu

# Navigate to frontend directory
cd frontend

# Install dependencies
npm install

# Start frontend development server
npm run dev