Spaces:
Sleeping
Sleeping
File size: 3,236 Bytes
aa4269d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | """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)}
|