refactor(tools): add BaseTools class and use XML prompts

sec_edgar_mcp/tools/__init__.py

@@ -1,7 +1,16 @@
+from .base import BaseTools, ToolResponse
 from .company import CompanyTools
 from .filings import FilingsTools
 from .financial import FinancialTools
 from .insider import InsiderTools
-from .types import ToolResponse
+from .xbrl import XBRLExtractor
 
-__all__ = ["CompanyTools", "FilingsTools", "FinancialTools", "InsiderTools", "ToolResponse"]
+__all__ = [
+    "BaseTools",
+    "CompanyTools",
+    "FilingsTools",
+    "FinancialTools",
+    "InsiderTools",
+    "ToolResponse",
+    "XBRLExtractor",
+]

sec_edgar_mcp/tools/base.py

@@ -0,0 +1,62 @@
+"""Base utilities for SEC EDGAR tools."""
+
+from datetime import date, datetime
+from typing import Any, Dict, Optional
+
+from ..core.client import EdgarClient
+
+ToolResponse = Dict[str, Any]
+
+
+class BaseTools:
+    """Base class with common utilities for all tool classes."""
+
+    def __init__(self):
+        self.client = EdgarClient()
+
+    def _parse_date(self, date_value) -> Optional[datetime]:
+        """Parse a date value to datetime."""
+        if date_value is None:
+            return None
+        if isinstance(date_value, datetime):
+            return date_value
+        if isinstance(date_value, date):
+            return datetime.combine(date_value, datetime.min.time())
+        if isinstance(date_value, str):
+            return datetime.fromisoformat(date_value.replace("Z", "+00:00"))
+        return None
+
+    def _format_date(self, date_value) -> str:
+        """Format a date value to ISO string."""
+        if hasattr(date_value, "isoformat"):
+            return date_value.isoformat()
+        return str(date_value)
+
+    def _find_filing(self, filings, accession_number: str):
+        """Find a filing by accession number."""
+        clean_accession = accession_number.replace("-", "")
+        for filing in filings:
+            if filing.accession_number.replace("-", "") == clean_accession:
+                return filing
+        return None
+
+    def _build_sec_url(self, cik: str, accession_number: str) -> str:
+        """Build SEC URL for a filing."""
+        clean_accession = accession_number.replace("-", "")
+        return f"https://www.sec.gov/Archives/edgar/data/{cik}/{clean_accession}/{accession_number}.txt"
+
+    def _create_filing_reference(
+        self, filing, cik: str, form_type: str, period_days: Optional[int] = None
+    ) -> Dict[str, Any]:
+        """Create a standard filing reference dict."""
+        ref: Dict[str, Any] = {
+            "filing_date": self._format_date(filing.filing_date),
+            "accession_number": filing.accession_number,
+            "form_type": form_type,
+            "sec_url": self._build_sec_url(cik, filing.accession_number),
+            "data_source": f"SEC EDGAR Filing {filing.accession_number}",
+            "disclaimer": "All data extracted directly from SEC EDGAR filing with exact precision.",
+        }
+        if period_days:
+            ref["period_analyzed"] = f"Last {period_days} days from {datetime.now().strftime('%Y-%m-%d')}"
+        return ref

sec_edgar_mcp/tools/company.py

@@ -1,36 +1,29 @@
-from ..core.client import EdgarClient
+"""Company-related tools for SEC EDGAR data."""
+
+from typing import Any, Dict
+
 from ..core.models import CompanyInfo
 from ..utils.exceptions import CompanyNotFoundError
-from .types import ToolResponse
-
+from .base import BaseTools, ToolResponse
 
-class CompanyTools:
-    """Tools for company-related operations."""
 
-    def __init__(self):
-        self.client = EdgarClient()
+class CompanyTools(BaseTools):
+    """Tools for retrieving company information from SEC EDGAR."""
 
     def get_cik_by_ticker(self, ticker: str) -> ToolResponse:
-        """Get the CIK for a company based on its ticker symbol."""
+        """Convert ticker symbol to CIK."""
         try:
             cik = self.client.get_cik_by_ticker(ticker)
             if cik:
-                return {
-                    "success": True,
-                    "cik": cik,
-                    "ticker": ticker.upper(),
-                    "suggestion": f"Use CIK '{cik}' instead of ticker '{ticker}' for more reliable and faster API calls",
-                }
-            else:
-                return {"success": False, "error": f"CIK not found for ticker: {ticker}"}
+                return {"success": True, "cik": cik, "ticker": ticker.upper()}
+            return {"success": False, "error": f"CIK not found for ticker: {ticker}"}
         except Exception as e:
             return {"success": False, "error": str(e)}
 
     def get_company_info(self, identifier: str) -> ToolResponse:
-        """Get detailed company information."""
+        """Get detailed company information from SEC records."""
         try:
             company = self.client.get_company(identifier)
-
             info = CompanyInfo(
                 cik=company.cik,
                 name=company.name,
@@ -41,81 +34,31 @@ def get_company_info(self, identifier: str) -> ToolResponse:
                 state=getattr(company, "state", None),
                 fiscal_year_end=getattr(company, "fiscal_year_end", None),
             )
-
             return {"success": True, "company": info.to_dict()}
         except CompanyNotFoundError as e:
             return {"success": False, "error": str(e)}
         except Exception as e:
-            return {"success": False, "error": f"Failed to get company info: {str(e)}"}
+            return {"success": False, "error": f"Failed to get company info: {e}"}
 
     def search_companies(self, query: str, limit: int = 10) -> ToolResponse:
         """Search for companies by name."""
         try:
             results = self.client.search_companies(query, limit)
-
-            companies = []
-            for result in results:
-                companies.append({"cik": result.cik, "name": result.name, "tickers": getattr(result, "tickers", [])})
-
+            companies = [{"cik": r.cik, "name": r.name, "tickers": getattr(r, "tickers", [])} for r in results]
             return {"success": True, "companies": companies, "count": len(companies)}
         except Exception as e:
-            return {"success": False, "error": f"Failed to search companies: {str(e)}"}
+            return {"success": False, "error": f"Failed to search companies: {e}"}
 
     def get_company_facts(self, identifier: str) -> ToolResponse:
-        """Get company facts and financial data."""
+        """Get company facts and financial data from XBRL."""
         try:
             company = self.client.get_company(identifier)
-
-            # Get company facts using edgar-tools
             facts = company.get_facts()
 
             if not facts:
                 return {"success": False, "error": "No facts available for this company"}
 
-            # Extract key financial metrics
-            metrics = {}
-
-            # Try to access the raw facts data
-            if hasattr(facts, "data"):
-                facts_data = facts.data
-
-                # Look for US-GAAP facts
-                if "us-gaap" in facts_data:
-                    gaap_facts = facts_data["us-gaap"]
-
-                    # Common metrics to extract
-                    metric_names = [
-                        "Assets",
-                        "Liabilities",
-                        "StockholdersEquity",
-                        "Revenues",
-                        "NetIncomeLoss",
-                        "EarningsPerShareBasic",
-                        "CashAndCashEquivalents",
-                        "CommonStockSharesOutstanding",
-                    ]
-
-                    for metric in metric_names:
-                        if metric in gaap_facts:
-                            metric_data = gaap_facts[metric]
-                            if "units" in metric_data:
-                                # Get the most recent value
-                                for unit_type, unit_data in metric_data["units"].items():
-                                    if unit_data:
-                                        # Sort by end date and get the latest
-                                        sorted_data = sorted(unit_data, key=lambda x: x.get("end", ""), reverse=True)
-                                        if sorted_data:
-                                            latest = sorted_data[0]
-                                            metrics[metric] = {
-                                                "value": float(latest.get("val", 0)),
-                                                "unit": unit_type,
-                                                "period": latest.get("end", ""),
-                                                "form": latest.get("form", ""),
-                                                "fiscal_year": latest.get("fy", ""),
-                                                "fiscal_period": latest.get("fp", ""),
-                                            }
-                                            break
-
+            metrics = self._extract_metrics(facts)
             return {
                 "success": True,
                 "cik": company.cik,
@@ -124,4 +67,54 @@ def get_company_facts(self, identifier: str) -> ToolResponse:
                 "has_facts": bool(facts),
             }
         except Exception as e:
-            return {"success": False, "error": f"Failed to get company facts: {str(e)}"}
+            return {"success": False, "error": f"Failed to get company facts: {e}"}
+
+    def _extract_metrics(self, facts) -> Dict[str, Any]:
+        """Extract key financial metrics from company facts."""
+        metrics: Dict[str, Any] = {}
+
+        if not hasattr(facts, "data"):
+            return metrics
+
+        facts_data = facts.data
+        if "us-gaap" not in facts_data:
+            return metrics
+
+        gaap_facts = facts_data["us-gaap"]
+        metric_names = [
+            "Assets",
+            "Liabilities",
+            "StockholdersEquity",
+            "Revenues",
+            "NetIncomeLoss",
+            "EarningsPerShareBasic",
+            "CashAndCashEquivalents",
+            "CommonStockSharesOutstanding",
+        ]
+
+        for metric in metric_names:
+            if metric not in gaap_facts:
+                continue
+
+            metric_data = gaap_facts[metric]
+            if "units" not in metric_data:
+                continue
+
+            for unit_type, unit_data in metric_data["units"].items():
+                if not unit_data:
+                    continue
+
+                sorted_data = sorted(unit_data, key=lambda x: x.get("end", ""), reverse=True)
+                if sorted_data:
+                    latest = sorted_data[0]
+                    metrics[metric] = {
+                        "value": float(latest.get("val", 0)),
+                        "unit": unit_type,
+                        "period": latest.get("end", ""),
+                        "form": latest.get("form", ""),
+                        "fiscal_year": latest.get("fy", ""),
+                        "fiscal_period": latest.get("fp", ""),
+                    }
+                    break
+
+        return metrics