MCP Stripe Server

# stripe_server/server.py (updated with logging) import os import json import logging from datetime import datetime from typing import Any, Sequence from functools import lru_cache import stripe from dotenv import load_dotenv import mcp.server.stdio from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions from mcp.types import Resource, Tool, TextContent from pydantic import AnyUrl from tools import get_stripe_tools load_dotenv() logging.basicConfig( level=logging.DEBUG if os.getenv("DEBUG") else logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger("stripe-mcp-server") def custom_json_serializer(obj): if isinstance(obj, datetime): return obj.isoformat() if isinstance(obj, stripe.StripeObject): return json.loads(str(obj)) raise TypeError(f"Object of type {type(obj)} is not JSON serializable") # Revised StripeManager class class StripeManager: def __init__(self): logger.info("🔄 Initializing StripeManager") self.audit_entries = [] # MUST be first line in __init__ stripe.api_key = os.getenv("STRIPE_API_KEY") if not stripe.api_key: logger.critical("❌ STRIPE_API_KEY missing") raise ValueError("STRIPE_API_KEY required") logger.info("✅ Stripe configured") logger.debug("Test connection...") try: # Verify API key works stripe.Customer.list(limit=1) except stripe.error.AuthenticationError as e: logger.critical("🔴 Invalid API key: %s", e) raise def log_operation(self, operation: str, parameters: dict) -> None: logger.debug("📝 Logging operation: %s with params: %s", operation, parameters) audit_entry = { "timestamp": datetime.utcnow().isoformat(), "operation": operation, "parameters": parameters } self.audit_entries.append(audit_entry) def _synthesize_audit_log(self) -> str: logger.debug("Generating audit log with %d entries", len(self.audit_entries)) if not self.audit_entries: return "No Stripe operations performed yet." report = "📋 Stripe Operations Audit Log 📋\n\n" for entry in self.audit_entries: report += f"[{entry['timestamp']}]\n" report += f"Operation: {entry['operation']}\n" report += f"Parameters: {json.dumps(entry['parameters'], indent=2)}\n" report += "-" * 50 + "\n" return report async def main(): logger.info("Starting Stripe MCP Server") manager = StripeManager() server = Server("stripe-mcp-server") @server.list_resources() async def handle_list_resources() -> list[Resource]: return [ Resource( uri=AnyUrl("audit://stripe-operations"), name="Stripe Operations Audit Log", description="Log of all Stripe operations performed", mimeType="text/plain", ) ] @server.read_resource() async def handle_read_resource(uri: AnyUrl) -> str: if uri.scheme != "audit": raise ValueError("Unsupported URI scheme") return manager._synthesize_audit_log() @server.list_tools() async def list_tools() -> list[Tool]: return get_stripe_tools() @server.call_tool() async def call_tool(name: str, arguments: Any) -> Sequence[TextContent]: logger.debug("=== RECEIVED JSON-RPC callTool request ===") try: if name.startswith("customer_"): return await handle_customer_operations(manager, name, arguments) elif name.startswith("payment_"): return await handle_payment_operations(manager, name, arguments) elif name.startswith("refund_"): return await handle_refund_operations(manager, name, arguments) else: raise ValueError(f"Unknown tool: {name}") except stripe.error.StripeError as e: logger.error(f"Stripe API error: {str(e)}") raise RuntimeError(f"Payment processing failed: {str(e)}") async def handle_customer_operations(manager, name: str, args: dict): if name == "customer_create": customer = stripe.Customer.create( email=args["email"], name=args.get("name"), metadata=args.get("metadata", {}) ) manager.log_operation("customer_create", args) # Include the "type" field along with "text" return [TextContent(type="text", text=json.dumps(customer, default=custom_json_serializer))] elif name == "customer_retrieve": customer = stripe.Customer.retrieve(args["customer_id"]) return [TextContent(type="text", text=json.dumps(customer, default=custom_json_serializer))] elif name == "customer_update": customer = stripe.Customer.modify( args["customer_id"], **args["update_fields"] ) manager.log_operation("customer_update", args) return [TextContent(type="text", text=json.dumps(customer, default=custom_json_serializer))] raise ValueError(f"Unknown customer operation: {name}") async def handle_payment_operations(manager, name: str, args: dict): if name == "payment_intent_create": intent = stripe.PaymentIntent.create( amount=args["amount"], currency=args["currency"], payment_method_types=args.get("payment_method_types", ["card"]), customer=args.get("customer"), metadata=args.get("metadata", {}) ) manager.log_operation("payment_intent_create", args) return [TextContent(type="text", text=json.dumps(intent, default=custom_json_serializer))] elif name == "charge_list": charges = stripe.Charge.list( limit=args.get("limit", 10), customer=args.get("customer_id") ) return [TextContent(type="text", text=json.dumps(charges, default=custom_json_serializer))] raise ValueError(f"Unknown payment operation: {name}") async def handle_refund_operations(manager, name: str, args: dict): if name == "refund_create": refund = stripe.Refund.create( charge=args["charge_id"], amount=args.get("amount"), reason=args.get("reason", "requested_by_customer") ) manager.log_operation("refund_create", args) return [TextContent(type="text", text=json.dumps(refund, default=custom_json_serializer))] raise ValueError(f"Unknown refund operation: {name}") async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, server.create_initialization_options(), ) if __name__ == "__main__": import asyncio asyncio.run(main())