"""MCP resources for statement catalog and data exposure."""
from typing import Optional
from ..database.sqlite_client import SQLiteClient
import json
class CatalogResource:
"""Resource provider for statement catalog."""
def __init__(self, db_client: SQLiteClient):
"""Initialize catalog resource.
Args:
db_client: SQLite database client
"""
self.db = db_client
async def get_catalog(self) -> str:
"""Get catalog of all statements.
Returns:
JSON string of statement catalog
"""
statements = await self.db.list_statements(limit=1000)
catalog = {
"total_statements": len(statements),
"statements": [
{
"statement_id": stmt["statement_id"],
"institution": stmt["institution"],
"account_number": stmt["account_number"],
"statement_date": stmt["statement_date"],
"indexed_at": stmt["indexed_at"],
}
for stmt in statements
],
}
return json.dumps(catalog, indent=2)
async def get_statement(self, statement_id: str) -> str:
"""Get specific statement data.
Args:
statement_id: Statement ID
Returns:
JSON string of statement data
"""
statement = await self.db.get_statement(statement_id)
if not statement:
return json.dumps({"error": "Statement not found"})
return json.dumps(statement, indent=2)
async def get_statement_summary(self, statement_id: str) -> str:
"""Get statement summary section.
Args:
statement_id: Statement ID
Returns:
JSON string of summary data
"""
statement = await self.db.get_statement(statement_id)
if not statement:
return json.dumps({"error": "Statement not found"})
# Extract summary fields from statement
summary = {
"statement_id": statement["statement_id"],
"account_number": statement["account_number"],
"institution": statement["institution"],
"statement_date": statement["statement_date"],
}
return json.dumps(summary, indent=2)
async def get_statement_holdings(self, statement_id: str) -> str:
"""Get holdings for a statement.
Args:
statement_id: Statement ID
Returns:
JSON string of holdings data
"""
# This would require a join or separate query
# For now, return placeholder
return json.dumps({"statement_id": statement_id, "holdings": []})
async def get_statement_transactions(self, statement_id: str) -> str:
"""Get transactions for a statement.
Args:
statement_id: Statement ID
Returns:
JSON string of transactions data
"""
# This would require a join or separate query
# For now, return placeholder
return json.dumps({"statement_id": statement_id, "transactions": []})
