Square MCP Server

Official
MIT License
Overview InspectNew Schema Related Servers Reviews Score
square-mcp
src
square_mcp
from typing import Dict, Any, Optional, List
from square.client import Client
import os
from datetime import datetime

from mcp.server.fastmcp import FastMCP
from mcp.shared.exceptions import McpError
from mcp.types import ErrorData, INTERNAL_ERROR, INVALID_PARAMS

# Initialize Square client
access_token = os.getenv('SQUARE_ACCESS_TOKEN')
environment = os.getenv('SQUARE_ENVIRONMENT', 'sandbox')  # Default to sandbox if not set

if not access_token:
    raise McpError(
        ErrorData(code=INVALID_PARAMS, message="SQUARE_ACCESS_TOKEN environment variable is required")
    )

if environment not in ['sandbox', 'production']:
    raise McpError(
        ErrorData(code=INVALID_PARAMS, message="SQUARE_ENVIRONMENT must be either 'sandbox' or 'production'")
    )

square_client = Client(
    access_token=access_token,
    environment=environment
)

mcp = FastMCP(
    "square",
    title="Square MCP",
    description="Square API Model Context Protocol Server",
    version="0.1.0",
)

@mcp.tool()
async def payments(
    operation: str,
    params: Dict[str, Any]
) -> Dict[str, Any]:
    """Manage payment operations using Square API

    Args:
        operation: The operation to perform. Valid operations:
            Payments:
                - list_payments
                - create_payment
                - get_payment
                - update_payment
                - cancel_payment
            Refunds:
                - refund_payment
                - list_refunds
                - get_refund
            Disputes:
                - list_disputes
                - retrieve_dispute
                - accept_dispute
                - create_dispute_evidence
            Gift Cards:
                - create_gift_card
                - link_customer_to_gift_card
                - retrieve_gift_card
                - list_gift_cards
            Bank Accounts:
                - list_bank_accounts
                - get_bank_account
        params: Dictionary of parameters for the specific operation
    """
    try:
        match operation:
            # Payments
            case "list_payments":
                result = square_client.payments.list_payments(**params)
            case "create_payment":
                result = square_client.payments.create_payment(params)
            case "get_payment":
                result = square_client.payments.get_payment(**params)
            case "update_payment":
                result = square_client.payments.update_payment(**params)
            case "cancel_payment":
                result = square_client.payments.cancel_payment(**params)
            # Refunds
            case "refund_payment":
                result = square_client.refunds.refund_payment(params)
            case "list_refunds":
                result = square_client.refunds.list_payment_refunds(**params)
            case "get_refund":
                result = square_client.refunds.get_payment_refund(**params)
            # Disputes
            case "list_disputes":
                result = square_client.disputes.list_disputes(**params)
            case "retrieve_dispute":
                result = square_client.disputes.retrieve_dispute(**params)
            case "accept_dispute":
                result = square_client.disputes.accept_dispute(**params)
            case "create_dispute_evidence":
                result = square_client.disputes.create_dispute_evidence(**params)
            # Gift Cards
            case "create_gift_card":
                result = square_client.gift_cards.create_gift_card(params)
            case "link_customer_to_gift_card":
                result = square_client.gift_cards.link_customer_to_gift_card(**params)
            case "retrieve_gift_card":
                result = square_client.gift_cards.retrieve_gift_card(**params)
            case "list_gift_cards":
                result = square_client.gift_cards.list_gift_cards(**params)
            # Bank Accounts
            case "list_bank_accounts":
                result = square_client.bank_accounts.list_bank_accounts(**params)
            case "get_bank_account":
                result = square_client.bank_accounts.get_bank_account(**params)
            case _:
                raise McpError(INVALID_PARAMS, ErrorData(message=f"Invalid operation: {operation}"))

        return result.body
    except Exception as e:
        raise McpError(INTERNAL_ERROR, ErrorData(message=str(e)))

@mcp.tool()
async def terminal(
    operation: str,
    params: Dict[str, Any]
) -> Dict[str, Any]:
    """Manage Square Terminal operations

    Args:
        operation: The operation to perform. Valid operations:
            Checkout:
                - create_terminal_checkout
                - search_terminal_checkouts
                - get_terminal_checkout
                - cancel_terminal_checkout
            Devices:
                - create_terminal_device
                - get_terminal_device
                - search_terminal_devices
            Refunds:
                - create_terminal_refund
                - search_terminal_refunds
                - get_terminal_refund
                - cancel_terminal_refund
        params: Dictionary of parameters for the specific operation
    """
    try:
        match operation:
            # Checkout
            case "create_terminal_checkout":
                result = square_client.terminal.create_terminal_checkout(params)
            case "search_terminal_checkouts":
                result = square_client.terminal.search_terminal_checkouts(params)
            case "get_terminal_checkout":
                result = square_client.terminal.get_terminal_checkout(**params)
            case "cancel_terminal_checkout":
                result = square_client.terminal.cancel_terminal_checkout(**params)
            # Devices
            case "create_terminal_device":
                result = square_client.terminal.create_terminal_device(params)
            case "get_terminal_device":
                result = square_client.terminal.get_terminal_device(**params)
            case "search_terminal_devices":
                result = square_client.terminal.search_terminal_devices(params)
            # Refunds
            case "create_terminal_refund":
                result = square_client.terminal.create_terminal_refund(params)
            case "search_terminal_refunds":
                result = square_client.terminal.search_terminal_refunds(params)
            case "get_terminal_refund":
                result = square_client.terminal.get_terminal_refund(**params)
            case "cancel_terminal_refund":
                result = square_client.terminal.cancel_terminal_refund(**params)
            case _:
                raise McpError(INVALID_PARAMS, ErrorData(message=f"Invalid operation: {operation}"))

        return result.body
    except Exception as e:
        raise McpError(INTERNAL_ERROR, ErrorData(message=str(e)))

@mcp.tool()
async def orders(
    operation: str,
    params: Dict[str, Any]
) -> Dict[str, Any]:
    """Manage orders and checkout operations

    Args:
        operation: The operation to perform. Valid operations:
            Orders:
                - create_order
                - batch_retrieve_orders
                - calculate_order
                - clone_order
                - search_orders
                - pay_order
                - update_order
            Checkout:
                - create_checkout
                - create_payment_link
            Custom Attributes:
                - upsert_order_custom_attribute
                - list_order_custom_attribute_definitions
        params: Dictionary of parameters for the specific operation
    """
    try:
        match operation:
            # Orders
            case "create_order":
                result = square_client.orders.create_order(params)
            case "batch_retrieve_orders":
                result = square_client.orders.batch_retrieve_orders(params)
            case "calculate_order":
                result = square_client.orders.calculate_order(params)
            case "clone_order":
                result = square_client.orders.clone_order(params)
            case "search_orders":
                result = square_client.orders.search_orders(params)
            case "pay_order":
                result = square_client.orders.pay_order(params)
            case "update_order":
                result = square_client.orders.update_order(**params)
            # Checkout
            case "create_checkout":
                result = square_client.checkout.create_checkout(params)
            case "create_payment_link":
                result = square_client.checkout.create_payment_link(params)
            # Custom Attributes
            case "upsert_order_custom_attribute":
                result = square_client.orders.upsert_order_custom_attribute(**params)
            case "list_order_custom_attribute_definitions":
                result = square_client.orders.list_order_custom_attribute_definitions(**params)
            case _:
                raise McpError(INVALID_PARAMS, ErrorData(message=f"Invalid operation: {operation}"))

        return result.body
    except Exception as e:
        raise McpError(INTERNAL_ERROR, ErrorData(message=str(e)))

@mcp.tool()
async def catalog(
    operation: str,
    params: Dict[str, Any]
) -> Dict[str, Any]:
    """Manage catalog operations

    Args:
        operation: The operation to perform. Valid operations:
            - create_catalog_object
            - batch_delete_catalog_objects
            - batch_retrieve_catalog_objects
            - batch_upsert_catalog_objects
            - create_catalog_image
            - delete_catalog_object
            - retrieve_catalog_object
            - search_catalog_objects
            - update_catalog_object
            - update_item_modifier_lists
            - update_item_taxes
        params: Dictionary of parameters for the specific operation
    """
    try:
        match operation:
            case "create_catalog_object":
                result = square_client.catalog.create_catalog_object(params)
            case "batch_delete_catalog_objects":
                result = square_client.catalog.batch_delete_catalog_objects(params)
            case "batch_retrieve_catalog_objects":
                result = square_client.catalog.batch_retrieve_catalog_objects(params)
            case "batch_upsert_catalog_objects":
                result = square_client.catalog.batch_upsert_catalog_objects(params)
            case "create_catalog_image":
                result = square_client.catalog.create_catalog_image(params)
            case "delete_catalog_object":
                result = square_client.catalog.delete_catalog_object(**params)
            case "retrieve_catalog_object":
                result = square_client.catalog.retrieve_catalog_object(**params)
            case "search_catalog_objects":
                result = square_client.catalog.search_catalog_objects(params)
            case "update_catalog_object":
                result = square_client.catalog.update_catalog_object(**params)
            case "update_item_modifier_lists":
                result = square_client.catalog.update_item_modifier_lists(params)
            case "update_item_taxes":
                result = square_client.catalog.update_item_taxes(params)
            case _:
                raise McpError(INVALID_PARAMS, ErrorData(message=f"Invalid operation: {operation}"))

        return result.body
    except Exception as e:
        raise McpError(INTERNAL_ERROR, ErrorData(message=str(e)))

@mcp.tool()
async def inventory(
    operation: str,
    params: Dict[str, Any]
) -> Dict[str, Any]:
    """Manage inventory operations

    Args:
        operation: The operation to perform. Valid operations:
            - batch_change_inventory
            - batch_retrieve_inventory_changes
            - batch_retrieve_inventory_counts
            - retrieve_inventory_adjustment
            - retrieve_inventory_changes
            - retrieve_inventory_count
            - retrieve_inventory_physical_count
            - retrieve_inventory_transfer
        params: Dictionary of parameters for the specific operation
    """
    try:
        match operation:
            case "batch_change_inventory":
                result = square_client.inventory.batch_change_inventory(params)
            case "batch_retrieve_inventory_changes":
                result = square_client.inventory.batch_retrieve_inventory_changes(params)
            case "batch_retrieve_inventory_counts":
                result = square_client.inventory.batch_retrieve_inventory_counts(params)
            case "retrieve_inventory_adjustment":
                result = square_client.inventory.retrieve_inventory_adjustment(**params)
            case "retrieve_inventory_changes":
                result = square_client.inventory.retrieve_inventory_changes(**params)
            case "retrieve_inventory_count":
                result = square_client.inventory.retrieve_inventory_count(**params)
            case "retrieve_inventory_physical_count":
                result = square_client.inventory.retrieve_inventory_physical_count(**params)
            case "retrieve_inventory_transfer":
                result = square_client.inventory.retrieve_inventory_transfer(**params)
            case _:
                raise McpError(INVALID_PARAMS, ErrorData(message=f"Invalid operation: {operation}"))

        return result.body
    except Exception as e:
        raise McpError(INTERNAL_ERROR, ErrorData(message=str(e)))

@mcp.tool()
async def subscriptions(
    operation: str,
    params: Dict[str, Any]
) -> Dict[str, Any]:
    """Manage subscription operations

    Args:
        operation: The operation to perform. Valid operations:
            - create_subscription
            - search_subscriptions
            - retrieve_subscription
            - update_subscription
            - cancel_subscription
            - list_subscription_events
            - pause_subscription
            - resume_subscription
            - swap_plan
        params: Dictionary of parameters for the specific operation
    """
    try:
        match operation:
            case "create_subscription":
                result = square_client.subscriptions.create_subscription(params)
            case "search_subscriptions":
                result = square_client.subscriptions.search_subscriptions(params)
            case "retrieve_subscription":
                result = square_client.subscriptions.retrieve_subscription(**params)
            case "update_subscription":
                result = square_client.subscriptions.update_subscription(**params)
            case "cancel_subscription":
                result = square_client.subscriptions.cancel_subscription(**params)
            case "list_subscription_events":
                result = square_client.subscriptions.list_subscription_events(**params)
            case "pause_subscription":
                result = square_client.subscriptions.pause_subscription(**params)
            case "resume_subscription":
                result = square_client.subscriptions.resume_subscription(**params)
            case "swap_plan":
                result = square_client.subscriptions.swap_plan(**params)
            case _:
                raise McpError(INVALID_PARAMS, ErrorData(message=f"Invalid operation: {operation}"))

        return result.body
    except Exception as e:
        raise McpError(INTERNAL_ERROR, ErrorData(message=str(e)))

@mcp.tool()
async def invoices(
    operation: str,
    params: Dict[str, Any]
) -> Dict[str, Any]:
    """Manage invoice operations

    Args:
        operation: The operation to perform. Valid operations:
            - create_invoice
            - search_invoices
            - get_invoice
            - update_invoice
            - cancel_invoice
            - publish_invoice
            - delete_invoice
        params: Dictionary of parameters for the specific operation
    """
    try:
        match operation:
            case "create_invoice":
                result = square_client.invoices.create_invoice(params)
            case "search_invoices":
                result = square_client.invoices.search_invoices(params)
            case "get_invoice":
                result = square_client.invoices.get_invoice(**params)
            case "update_invoice":
                result = square_client.invoices.update_invoice(**params)
            case "cancel_invoice":
                result = square_client.invoices.cancel_invoice(**params)
            case "publish_invoice":
                result = square_client.invoices.publish_invoice(**params)
            case "delete_invoice":
                result = square_client.invoices.delete_invoice(**params)
            case _:
                raise McpError(INVALID_PARAMS, ErrorData(message=f"Invalid operation: {operation}"))

        return result.body
    except Exception as e:
        raise McpError(INTERNAL_ERROR, ErrorData(message=str(e)))

@mcp.tool()
async def team(
    operation: str,
    params: Dict[str, Any]
) -> Dict[str, Any]:
    """Manage team operations

    Args:
        operation: The operation to perform. Valid operations:
            Team Members:
                - create_team_member
                - bulk_create_team_members
                - update_team_member
                - retrieve_team_member
                - search_team_members
            Wages:
                - retrieve_wage_setting
                - update_wage_setting
            Labor:
                - create_break_type
                - create_shift
                - search_shifts
                - update_shift
                - create_workweek_config
            Cash Drawers:
                - list_cash_drawer_shifts
                - retrieve_cash_drawer_shift
        params: Dictionary of parameters for the specific operation
    """
    try:
        match operation:
            # Team Members
            case "create_team_member":
                result = square_client.team.create_team_member(params)
            case "bulk_create_team_members":
                result = square_client.team.bulk_create_team_members(params)
            case "update_team_member":
                result = square_client.team.update_team_member(**params)
            case "retrieve_team_member":
                result = square_client.team.retrieve_team_member(**params)
            case "search_team_members":
                result = square_client.team.search_team_members(params)
            # Wages
            case "retrieve_wage_setting":
                result = square_client.labor.retrieve_wage_setting(**params)
            case "update_wage_setting":
                result = square_client.labor.update_wage_setting(**params)
            # Labor
            case "create_break_type":
                result = square_client.labor.create_break_type(params)
            case "create_shift":
                result = square_client.labor.create_shift(params)
            case "search_shifts":
                result = square_client.labor.search_shifts(params)
            case "update_shift":
                result = square_client.labor.update_shift(**params)
            case "create_workweek_config":
                result = square_client.labor.create_workweek_config(params)
            # Cash Drawers
            case "list_cash_drawer_shifts":
                result = square_client.cash_drawers.list_cash_drawer_shifts(**params)
            case "retrieve_cash_drawer_shift":
                result = square_client.cash_drawers.retrieve_cash_drawer_shift(**params)
            case _:
                raise McpError(INVALID_PARAMS, ErrorData(message=f"Invalid operation: {operation}"))

        return result.body
    except Exception as e:
        raise McpError(INTERNAL_ERROR, ErrorData(message=str(e)))

@mcp.tool()
async def customers(
    operation: str,
    params: Dict[str, Any]
) -> Dict[str, Any]:
    """Manage customer operations

    Args:
        operation: The operation to perform. Valid operations:
            Customers:
                - list_customers
                - create_customer
                - delete_customer
                - retrieve_customer
                - update_customer
                - search_customers
            Groups:
                - create_customer_group
                - delete_customer_group
                - list_customer_groups
                - retrieve_customer_group
                - update_customer_group
            Segments:
                - list_customer_segments
                - retrieve_customer_segment
            Custom Attributes:
                - create_customer_custom_attribute_definition
                - delete_customer_custom_attribute_definition
                - list_customer_custom_attribute_definitions
        params: Dictionary of parameters for the specific operation
    """
    try:
        match operation:
            # Customers
            case "list_customers":
                result = square_client.customers.list_customers(**params)
            case "create_customer":
                result = square_client.customers.create_customer(params)
            case "delete_customer":
                result = square_client.customers.delete_customer(**params)
            case "retrieve_customer":
                result = square_client.customers.retrieve_customer(**params)
            case "update_customer":
                result = square_client.customers.update_customer(**params)
            case "search_customers":
                result = square_client.customers.search_customers(params)
            # Groups
            case "create_customer_group":
                result = square_client.customer_groups.create_customer_group(params)
            case "delete_customer_group":
                result = square_client.customer_groups.delete_customer_group(**params)
            case "list_customer_groups":
                result = square_client.customer_groups.list_customer_groups(**params)
            case "retrieve_customer_group":
                result = square_client.customer_groups.retrieve_customer_group(**params)
            case "update_customer_group":
                result = square_client.customer_groups.update_customer_group(**params)
            # Segments
            case "list_customer_segments":
                result = square_client.customer_segments.list_customer_segments(**params)
            case "retrieve_customer_segment":
                result = square_client.customer_segments.retrieve_customer_segment(**params)
            # Custom Attributes
            case "create_customer_custom_attribute_definition":
                result = square_client.customer_custom_attributes.create_customer_custom_attribute_definition(params)
            case "delete_customer_custom_attribute_definition":
                result = square_client.customer_custom_attributes.delete_customer_custom_attribute_definition(**params)
            case "list_customer_custom_attribute_definitions":
                result = square_client.customer_custom_attributes.list_customer_custom_attribute_definitions(**params)
            case _:
                raise McpError(INVALID_PARAMS, ErrorData(message=f"Invalid operation: {operation}"))

        return result.body
    except Exception as e:
        raise McpError(INTERNAL_ERROR, ErrorData(message=str(e)))

@mcp.tool()
async def loyalty(
    operation: str,
    params: Dict[str, Any]
) -> Dict[str, Any]:
    """Manage loyalty operations

    Args:
        operation: The operation to perform. Valid operations:
            Programs:
                - create_loyalty_program
                - retrieve_loyalty_program
            Accounts:
                - create_loyalty_account
                - search_loyalty_accounts
                - retrieve_loyalty_account
                - accumulate_loyalty_points
                - adjust_loyalty_points
                - search_loyalty_events
            Promotions:
                - create_loyalty_promotion
                - cancel_loyalty_promotion
        params: Dictionary of parameters for the specific operation
    """
    try:
        match operation:
            # Programs
            case "create_loyalty_program":
                result = square_client.loyalty.create_loyalty_program(params)
            case "retrieve_loyalty_program":
                result = square_client.loyalty.retrieve_loyalty_program(**params)
            # Accounts
            case "create_loyalty_account":
                result = square_client.loyalty.create_loyalty_account(params)
            case "search_loyalty_accounts":
                result = square_client.loyalty.search_loyalty_accounts(params)
            case "retrieve_loyalty_account":
                result = square_client.loyalty.retrieve_loyalty_account(**params)
            case "accumulate_loyalty_points":
                result = square_client.loyalty.accumulate_loyalty_points(**params)
            case "adjust_loyalty_points":
                result = square_client.loyalty.adjust_loyalty_points(**params)
            case "search_loyalty_events":
                result = square_client.loyalty.search_loyalty_events(params)
            # Promotions
            case "create_loyalty_promotion":
                result = square_client.loyalty.create_loyalty_promotion(**params)
            case "cancel_loyalty_promotion":
                result = square_client.loyalty.cancel_loyalty_promotion(**params)
            case _:
                raise McpError(INVALID_PARAMS, ErrorData(message=f"Invalid operation: {operation}"))

        return result.body
    except Exception as e:
        raise McpError(INTERNAL_ERROR, ErrorData(message=str(e)))

@mcp.tool()
async def bookings(
    operation: str,
    params: Dict[str, Any]
) -> Dict[str, Any]:
    """Manage booking operations

    Args:
        operation: The operation to perform. Valid operations:
            Bookings:
                - create_booking
                - search_bookings
                - retrieve_booking
                - update_booking
                - cancel_booking
            Team Member Bookings:
                - bulk_retrieve_team_member_bookings
                - retrieve_team_member_booking_profile
            Location Profiles:
                - list_location_booking_profiles
                - retrieve_location_booking_profile
            Custom Attributes:
                - create_booking_custom_attribute_definition
                - update_booking_custom_attribute_definition
        params: Dictionary of parameters for the specific operation
    """
    try:
        match operation:
            # Bookings
            case "create_booking":
                result = square_client.bookings.create_booking(params)
            case "search_bookings":
                result = square_client.bookings.search_bookings(params)
            case "retrieve_booking":
                result = square_client.bookings.retrieve_booking(**params)
            case "update_booking":
                result = square_client.bookings.update_booking(**params)
            case "cancel_booking":
                result = square_client.bookings.cancel_booking(**params)
            # Team Member Bookings
            case "bulk_retrieve_team_member_bookings":
                result = square_client.bookings.bulk_retrieve_team_member_bookings(params)
            case "retrieve_team_member_booking_profile":
                result = square_client.bookings.retrieve_team_member_booking_profile(**params)
            # Location Profiles
            case "list_location_booking_profiles":
                result = square_client.bookings.list_location_booking_profiles(**params)
            case "retrieve_location_booking_profile":
                result = square_client.bookings.retrieve_location_booking_profile(**params)
            # Custom Attributes
            case "create_booking_custom_attribute_definition":
                result = square_client.booking_custom_attributes.create_booking_custom_attribute_definition(params)
            case "update_booking_custom_attribute_definition":
                result = square_client.booking_custom_attributes.update_booking_custom_attribute_definition(**params)
            case _:
                raise McpError(INVALID_PARAMS, ErrorData(message=f"Invalid operation: {operation}"))

        return result.body
    except Exception as e:
        raise McpError(INTERNAL_ERROR, ErrorData(message=str(e)))

@mcp.tool()
async def business(
    operation: str,
    params: Dict[str, Any]
) -> Dict[str, Any]:
    """Manage business operations

    Args:
        operation: The operation to perform. Valid operations:
            Merchants:
                - list_merchants
                - retrieve_merchant
            Locations:
                - list_locations
                - create_location
                - retrieve_location
                - update_location
            Vendors:
                - bulk_create_vendors
                - bulk_retrieve_vendors
                - create_vendor
                - search_vendors
                - update_vendor
            Sites:
                - list_sites
        params: Dictionary of parameters for the specific operation
    """
    try:
        match operation:
            # Merchants
            case "list_merchants":
                result = square_client.merchants.list_merchants(**params)
            case "retrieve_merchant":
                result = square_client.merchants.retrieve_merchant(**params)
            # Locations
            case "list_locations":
                result = square_client.locations.list_locations()
            case "create_location":
                result = square_client.locations.create_location(params)
            case "retrieve_location":
                result = square_client.locations.retrieve_location(**params)
            case "update_location":
                result = square_client.locations.update_location(**params)
            # Vendors
            case "bulk_create_vendors":
                result = square_client.vendors.bulk_create_vendors(params)
            case "bulk_retrieve_vendors":
                result = square_client.vendors.bulk_retrieve_vendors(params)
            case "create_vendor":
                result = square_client.vendors.create_vendor(params)
            case "search_vendors":
                result = square_client.vendors.search_vendors(params)
            case "update_vendor":
                result = square_client.vendors.update_vendor(**params)
            # Sites
            case "list_sites":
                result = square_client.sites.list_sites(**params)
            case _:
                raise McpError(INVALID_PARAMS, ErrorData(message=f"Invalid operation: {operation}"))

        return result.body
    except Exception as e:
        raise McpError(INTERNAL_ERROR, ErrorData(message=str(e)))