politics-tracking/backend/politrack/mcp_server.py

"""MCP (Model Context Protocol) server so AI agents can query Politrack with citations.

Connect an MCP client (Claude Desktop/Code, etc.) to http://<host>:8001/mcp
(streamable HTTP transport). All tools return evidence URLs; agents should cite them.
Tools are thin wrappers around politrack.queries - the same functions the REST API uses.
"""

import json
from datetime import date

from mcp.server.fastmcp import FastMCP

from politrack import queries
from politrack.db import connection
from politrack.queries import SourceFilter

mcp = FastMCP(
    "politrack",
    instructions=(
        "Evidence-backed data on Austrian politicians and parties. Every claim has an `evidence` list of URLs: "
        "always cite them. status 'corroborated' = confirmed by at least two independent sources. "
        "Claims of type legal_proceeding: unless legal_stage is 'convicted_final' (and the claim is not "
        "'disputed'), the presumption of innocence applies - always state 'Es gilt die Unschuldsvermutung' / "
        "'presumed innocent'. Parliamentary votes in Austria are mostly recorded per faction; roll_call=true "
        "votes contain individual positions. Travel/meeting dates are publication dates of the source. "
        "Money flows (donations, public contracts, government ads, party funding) come from official registers; "
        "a flow to an organisation connected to a person is a connection, not an accusation. "
        "Social accounts marked verified=false come from Wikidata and were not confirmed by an official source."
    ),
)


def _dump(rows) -> str:
    return json.dumps(rows, default=str, ensure_ascii=False)


def _date(value: str | None) -> date | None:
    return date.fromisoformat(value) if value else None


@mcp.tool()
def search(query: str, limit: int = 10, since: str | None = None, until: str | None = None,
           exclude_sources: list[str] | None = None) -> str:
    """Hybrid (keyword + semantic) search over all claims and documents. Dates as YYYY-MM-DD."""
    with connection() as conn:
        return _dump(queries.search(conn, query, sources=SourceFilter(exclude_sources or []), since=_date(since),
                                    until=_date(until), limit=limit))


@mcp.tool()
def find_person(name: str, status: str = "all") -> str:
    """Find politicians by (part of) their name. status: active | former | party_member | deceased | all."""
    with connection() as conn:
        return _dump(queries.list_persons(conn, q=name, status=None if status == "all" else status, limit=20))


@mcp.tool()
def get_person(slug: str) -> str:
    """Profile: activity status, roles/mandates over time (with sources), social accounts, claim counts."""
    with connection() as conn:
        return _dump(queries.get_person(conn, slug))


@mcp.tool()
def person_network(slug: str) -> str:
    """Organisations connected to a person (side jobs, career, party, meetings, lobby contacts) with the
    money those organisations paid or received (donations, public contracts, government ads)."""
    with connection() as conn:
        return _dump(queries.person_network(conn, slug))


@mcp.tool()
def list_parties() -> str:
    """All parties with number of currently active officials."""
    with connection() as conn:
        return _dump(queries.list_parties(conn))


@mcp.tool()
def get_party(slug: str) -> str:
    """Party overview: current officials, former members, money per year, finance reports, votes, dissenters,
    affiliated organisations and holdings."""
    with connection() as conn:
        return _dump(queries.get_party(conn, slug))


@mcp.tool()
def find_organization(name: str, kind: str | None = None) -> str:
    """Find organisations (companies, associations, public bodies, media, lobby groups) by name.
    kind: party | company | association | public_body | ministry | media | ngo | eu_institution | other."""
    with connection() as conn:
        return _dump(queries.list_organizations(conn, q=name, kind=kind, limit=20))


@mcp.tool()
def get_organization(slug: str) -> str:
    """Organisation profile: identifiers (Firmenbuch, UID, EU transparency register), money paid/received by
    kind, top counterparties and connected people."""
    with connection() as conn:
        return _dump(queries.get_organization(conn, slug))


@mcp.tool()
def list_claims(
    person: str | None = None,
    party: str | None = None,
    organization: str | None = None,
    claim_types: list[str] | None = None,
    topics: list[str] | None = None,
    since: str | None = None,
    until: str | None = None,
    limit: int = 30,
    exclude_sources: list[str] | None = None,
    min_trust: int = 0,
) -> str:
    """Claims about a person/party/organisation, newest first. claim_types: role, vote, donation, money_flow,
    statement, appearance, meeting, travel, side_job, legal_proceeding, lobbying, media_mention, career,
    parliamentary_question, government_decision, wiki_edit, declaration, ownership, affiliation, party_finance.
    Dates as YYYY-MM-DD."""
    with connection() as conn:
        return _dump(queries.list_claims(
            conn, sources=SourceFilter(exclude_sources or [], min_trust), person=person, party=party,
            organization=organization, claim_types=claim_types, topics=topics, since=_date(since),
            until=_date(until), limit=limit))


@mcp.tool()
def get_claim(claim_id: int) -> str:
    """One claim with all its evidence, e.g. from a permalink (/angaben/<id>) a user shared."""
    with connection() as conn:
        rows = queries.list_claims(conn, sources=SourceFilter([], 0), claim_id=claim_id, limit=1)
    return _dump(rows[0] if rows else {"error": f"claim {claim_id} not found"})


@mcp.tool()
def money_flows(kinds: list[str] | None = None, organization: str | None = None, person: str | None = None,
                query: str | None = None, since: str | None = None, until: str | None = None,
                min_amount: float | None = None, limit: int = 30) -> str:
    """Money flows from official registers. kinds: donation, contract, government_ad, political_ad, subsidy,
    party_funding, party_ad, loan, side_income, sponsorship. `person` = flows of organisations connected to
    that person. Dates as YYYY-MM-DD."""
    with connection() as conn:
        return _dump(queries.list_money(conn, sources=SourceFilter(), kinds=kinds, org=organization, person=person,
                                        q=query, since=_date(since), until=_date(until), min_amount=min_amount,
                                        limit=limit))


@mcp.tool()
def social_posts(person: str | None = None, party: str | None = None, platforms: list[str] | None = None,
                 limit: int = 30) -> str:
    """Public posts from politicians' accounts. platforms: bluesky, mastodon, telegram, youtube."""
    with connection() as conn:
        return _dump(queries.social_feed(conn, sources=SourceFilter(), person=person, party=party,
                                         platforms=platforms, limit=limit))


@mcp.tool()
def wikipedia_edits(limit: int = 50) -> str:
    """Wikipedia edits made from IP ranges of parliament/government (anonymous edits only)."""
    with connection() as conn:
        return _dump(queries.wiki_edits(conn, limit=limit))


@mcp.tool()
def calendar(start: str, end: str, person: str | None = None, party: str | None = None) -> str:
    """Events (sessions, hearings, council of ministers, trips, meetings) and per-day activity between two dates."""
    with connection() as conn:
        return _dump(queries.calendar(conn, sources=SourceFilter(), start=date.fromisoformat(start),
                                      end=date.fromisoformat(end), person=person, party=party))


@mcp.tool()
def person_votes(slug: str, topics: list[str] | None = None, limit: int = 50) -> str:
    """Votes: individual roll-call votes (with dissent flag) and faction votes during the person's mandates.
    `spoke` links a speech by the person on the same item (say vs. vote)."""
    with connection() as conn:
        return _dump(queries.person_votes(conn, slug, topics=topics, limit=limit))


@mcp.tool()
def former_politicians(party: str | None = None, limit: int = 30) -> str:
    """Former politicians with their latest news coverage and career entries."""
    with connection() as conn:
        return _dump(queries.former_overview(conn, party=party, limit=limit))


@mcp.tool()
def list_topics() -> str:
    """Topic taxonomy (slugs usable as filters)."""
    with connection() as conn:
        return _dump(queries.list_topics(conn))


def serve(host: str, port: int) -> None:
    mcp.settings.host = host
    mcp.settings.port = port
    mcp.run(transport="streamable-http")