mcp_server.py

"""
MCP Server for SEC Financial Report Data Query
Based on FastAPI framework
"""

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
from edgar_client import EdgarDataClient
import uvicorn

# Initialize FastAPI app
app = FastAPI(
    title="SEC Financial Report MCP Server",
    description="MCP Server for querying SEC EDGAR financial data",
    version="1.0.0"
)

# Configure CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Initialize EDGAR client with user information
edgar_client = EdgarDataClient(
    user_agent="Juntao Peng Financial Report Metrics App (jtyxabc@gmail.com)"
)


# Request/Response Models
class UserAgentRequest(BaseModel):
    user_agent: str = Field(
        default="Financial Report Metrics App (your-email@example.com)",
        description="User agent string for identifying request source"
    )


class CompanySearchRequest(BaseModel):
    company_name: str = Field(..., description="Company name to search")


class CompanyInfoRequest(BaseModel):
    cik: str = Field(..., description="Company CIK code")


class CompanyFilingsRequest(BaseModel):
    cik: str = Field(..., description="Company CIK code")
    form_types: Optional[List[str]] = Field(
        None, 
        description="List of form types (e.g., ['10-K', '10-Q']), None for all types"
    )


class CompanyFactsRequest(BaseModel):
    cik: str = Field(..., description="Company CIK code")


class FinancialDataRequest(BaseModel):
    cik: str = Field(..., description="Company CIK code")
    period: str = Field(
        ..., 
        description="Period in format 'YYYY' or 'YYYYQX' (e.g., '2025' or '2025Q3')"
    )


class CompanySearchResponse(BaseModel):
    cik: Optional[str] = None
    name: Optional[str] = None
    ticker: Optional[str] = None
    error: Optional[str] = None


class CompanyInfoResponse(BaseModel):
    cik: Optional[str] = None
    name: Optional[str] = None
    tickers: Optional[List[str]] = None
    sic: Optional[str] = None
    sic_description: Optional[str] = None
    error: Optional[str] = None


class FilingItem(BaseModel):
    form_type: str
    filing_date: str
    accession_number: str
    primary_document: str


class CompanyFilingsResponse(BaseModel):
    filings: List[FilingItem] = []
    error: Optional[str] = None


class CompanyFactsResponse(BaseModel):
    facts: Dict[str, Any] = {}
    error: Optional[str] = None


class FinancialDataResponse(BaseModel):
    period: Optional[str] = None
    total_revenue: Optional[float] = None
    net_income: Optional[float] = None
    earnings_per_share: Optional[float] = None
    operating_expenses: Optional[float] = None
    operating_cash_flow: Optional[float] = None
    source_url: Optional[str] = None
    source_form: Optional[str] = None
    data_source: Optional[str] = None
    additional_details: Optional[Dict[str, Any]] = None
    error: Optional[str] = None


# API Endpoints
@app.get("/")
async def root():
    """Root endpoint with API information"""
    return {
        "service": "SEC Financial Report MCP Server",
        "version": "1.0.0",
        "description": "MCP Server for querying SEC EDGAR financial data",
        "endpoints": {
            "health": "/health",
            "search_company": "/api/search_company",
            "get_company_info": "/api/get_company_info",
            "get_company_filings": "/api/get_company_filings",
            "get_company_facts": "/api/get_company_facts",
            "get_financial_data": "/api/get_financial_data"
        },
        "user_agent": "Juntao Peng (jtyxabc@gmail.com)"
    }


@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "service": "SEC Financial Report MCP Server"}


@app.post("/api/search_company", response_model=CompanySearchResponse)
async def search_company_by_name(request: CompanySearchRequest):
    """
    Search company CIK by company name
    
    Args:
        company_name (str): Company name
        
    Returns:
        dict: Dictionary containing company information
    """
    try:
        result = edgar_client.search_company_by_name(request.company_name)
        
        if result is None:
            return CompanySearchResponse(
                error=f"No company found matching '{request.company_name}'"
            )
        
        return CompanySearchResponse(
            cik=result.get("cik"),
            name=result.get("name"),
            ticker=result.get("ticker")
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/get_company_info", response_model=CompanyInfoResponse)
async def get_company_info(request: CompanyInfoRequest):
    """
    Get basic company information
    
    Args:
        cik (str): Company CIK code
        
    Returns:
        dict: Dictionary containing company information
    """
    try:
        result = edgar_client.get_company_info(request.cik)
        
        if result is None:
            return CompanyInfoResponse(
                error=f"No company information found for CIK: {request.cik}"
            )
        
        return CompanyInfoResponse(
            cik=result.get("cik"),
            name=result.get("name"),
            tickers=result.get("tickers"),
            sic=result.get("sic"),
            sic_description=result.get("sic_description")
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/get_company_filings", response_model=CompanyFilingsResponse)
async def get_company_filings(request: CompanyFilingsRequest):
    """
    Get all company filing documents
    
    Args:
        cik (str): Company CIK code
        form_types (list): List of form types (e.g., ['10-K', '10-Q']), None for all types
        
    Returns:
        list: List of filing documents
    """
    try:
        result = edgar_client.get_company_filings(request.cik, request.form_types)
        
        if not result:
            return CompanyFilingsResponse(
                filings=[],
                error=f"No filings found for CIK: {request.cik}"
            )
        
        filings = [
            FilingItem(
                form_type=filing.get("form_type", ""),
                filing_date=filing.get("filing_date", ""),
                accession_number=filing.get("accession_number", ""),
                primary_document=filing.get("primary_document", "")
            )
            for filing in result
        ]
        
        return CompanyFilingsResponse(filings=filings)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/get_company_facts", response_model=CompanyFactsResponse)
async def get_company_facts(request: CompanyFactsRequest):
    """
    Get all company financial facts data
    
    Args:
        cik (str): Company CIK code
        
    Returns:
        dict: Company financial facts data
    """
    try:
        result = edgar_client.get_company_facts(request.cik)
        
        if not result:
            return CompanyFactsResponse(
                facts={},
                error=f"No financial facts found for CIK: {request.cik}"
            )
        
        return CompanyFactsResponse(facts=result)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/get_financial_data", response_model=FinancialDataResponse)
async def get_financial_data_for_period(request: FinancialDataRequest):
    """
    Get financial data for a specific period (supports annual and quarterly)
    
    Args:
        cik (str): Company CIK code
        period (str): Period in format 'YYYY' or 'YYYYQX' (e.g., '2025' or '2025Q3')
        
    Returns:
        dict: Financial data dictionary
    """
    try:
        result = edgar_client.get_financial_data_for_period(request.cik, request.period)
        
        if not result or "period" not in result:
            return FinancialDataResponse(
                error=f"No financial data found for CIK: {request.cik}, Period: {request.period}"
            )
        
        # Collect additional details
        additional_details = {}
        for key, value in result.items():
            if key.endswith("_details"):
                additional_details[key] = value
        
        return FinancialDataResponse(
            period=result.get("period"),
            total_revenue=result.get("total_revenue"),
            net_income=result.get("net_income"),
            earnings_per_share=result.get("earnings_per_share"),
            operating_expenses=result.get("operating_expenses"),
            operating_cash_flow=result.get("operating_cash_flow"),
            source_url=result.get("source_url"),
            source_form=result.get("source_form"),
            data_source=result.get("data_source"),
            additional_details=additional_details if additional_details else None
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


if __name__ == "__main__":
    # Run the server
    uvicorn.run(
        "mcp_server:app",
        host="0.0.0.0",
        port=7860,
        reload=True
    )