src/tools/external.py

"""External data tools: SEC EDGAR, Companies House, currency conversion.

Thin, dependency-light wrappers. Each returns plain dicts so the same
functions back both the MCP server and the in-process LangChain tools.
"""

from __future__ import annotations

import requests

from config import COMPANIES_HOUSE_API_KEY, SEC_EDGAR_USER_AGENT

_TIMEOUT = 20


def sec_edgar_search(company: str, form_type: str = "10-K", limit: int = 5) -> dict:
    """Search recent SEC filings for a company via EDGAR full-text search."""
    try:
        resp = requests.get(
            "https://efts.sec.gov/LATEST/search-index",
            params={"q": company, "forms": form_type},
            headers={"User-Agent": SEC_EDGAR_USER_AGENT},
            timeout=_TIMEOUT,
        )
        if resp.status_code != 200:
            # fall back to the public full-text search endpoint
            resp = requests.get(
                "https://efts.sec.gov/LATEST/search-index?q=%22{}%22".format(company),
                headers={"User-Agent": SEC_EDGAR_USER_AGENT}, timeout=_TIMEOUT,
            )
        resp.raise_for_status()
        hits = resp.json().get("hits", {}).get("hits", [])[:limit]
        return {"source": "SEC EDGAR", "company": company, "results": [
            {"form": h["_source"].get("form"), "filed": h["_source"].get("file_date"),
             "name": h["_source"].get("display_names")}
            for h in hits
        ]}
    except Exception as e:  # network optional — never crash the agent loop
        return {"source": "SEC EDGAR", "error": str(e)}


def companies_house_search(company: str, limit: int = 5) -> dict:
    """Search UK Companies House (requires COMPANIES_HOUSE_API_KEY)."""
    if not COMPANIES_HOUSE_API_KEY:
        return {"source": "Companies House", "error": "COMPANIES_HOUSE_API_KEY not set"}
    try:
        resp = requests.get(
            "https://api.company-information.service.gov.uk/search/companies",
            params={"q": company, "items_per_page": limit},
            auth=(COMPANIES_HOUSE_API_KEY, ""),
            timeout=_TIMEOUT,
        )
        resp.raise_for_status()
        items = resp.json().get("items", [])
        return {"source": "Companies House", "company": company, "results": [
            {"title": i.get("title"), "company_number": i.get("company_number"),
             "status": i.get("company_status"), "incorporated": i.get("date_of_creation")}
            for i in items
        ]}
    except Exception as e:
        return {"source": "Companies House", "error": str(e)}


def convert_currency(amount: float, from_currency: str, to_currency: str) -> dict:
    """Convert currency using the free frankfurter.app ECB-rate API."""
    try:
        resp = requests.get(
            "https://api.frankfurter.app/latest",
            params={"amount": amount, "from": from_currency.upper(), "to": to_currency.upper()},
            timeout=_TIMEOUT,
        )
        resp.raise_for_status()
        data = resp.json()
        return {"amount": amount, "from": from_currency.upper(), "to": to_currency.upper(),
                "converted": data["rates"].get(to_currency.upper()), "date": data.get("date")}
    except Exception as e:
        return {"error": str(e)}