monarch-mcp

Instructions

Implementation Reference

• src/monarch_mcp/server.py:232-258 (handler)

  The get_accounts tool handler function that retrieves all financial accounts from Monarch Money. It uses an async helper to call the MonarchMoney client's get_accounts() method, formats the response into a clean list with id, name, type, balance, institution, and is_active fields, and returns JSON.

  @mcp.tool()
  @_handle_mcp_errors("getting accounts")
  def get_accounts() -> str:
      """Get all financial accounts from Monarch Money."""
  
      async def _get_accounts():
          client = await get_monarch_client()
          return await client.get_accounts()
  
      accounts = run_async(_get_accounts())
  
      # Format accounts for display
      account_list = []
      for account in accounts.get("accounts", []):
          account_info = {
              "id": account.get("id"),
              "name": account.get("displayName") or account.get("name"),
              "type": (account.get("type") or {}).get("name"),
              "balance": account.get("currentBalance"),
              "institution": (account.get("institution") or {}).get("name"),
              "is_active": account.get("isActive")
              if "isActive" in account
              else not account.get("deactivatedAt"),
          }
          account_list.append(account_info)
  
      return json.dumps(account_list, indent=2, default=str)

• src/monarch_mcp/server.py:232-233 (registration)

  Registration of the get_accounts tool with the MCP server using the @mcp.tool() decorator from FastMCP. The decorator automatically registers the function as an available tool.

  @mcp.tool()
  @_handle_mcp_errors("getting accounts")

• src/monarch_mcp/server.py:234-258 (schema)

  The schema is implicit in the function signature and return formatting. The function takes no parameters and returns a JSON string containing a list of account objects with fields: id, name, type, balance, institution, and is_active. No formal validation is performed, but the output structure is consistent.

  def get_accounts() -> str:
      """Get all financial accounts from Monarch Money."""
  
      async def _get_accounts():
          client = await get_monarch_client()
          return await client.get_accounts()
  
      accounts = run_async(_get_accounts())
  
      # Format accounts for display
      account_list = []
      for account in accounts.get("accounts", []):
          account_info = {
              "id": account.get("id"),
              "name": account.get("displayName") or account.get("name"),
              "type": (account.get("type") or {}).get("name"),
              "balance": account.get("currentBalance"),
              "institution": (account.get("institution") or {}).get("name"),
              "is_active": account.get("isActive")
              if "isActive" in account
              else not account.get("deactivatedAt"),
          }
          account_list.append(account_info)
  
      return json.dumps(account_list, indent=2, default=str)

• src/monarch_mcp/server.py:82-118 (helper)

  Error handling decorator that wraps the get_accounts tool to provide comprehensive exception handling. It catches RuntimeError, TransportServerError, TransportQueryError, TransportError, and generic exceptions, returning user-friendly error messages.

  def _handle_mcp_errors(operation: str):
      """Decorator providing granular exception handling for MCP tool functions.
  
      Catches specific known exception types with appropriate log messages,
      with a catch-all for anything unexpected.  Every path returns a
      user-readable error string so the MCP tool never crashes.
      """
      def decorator(func):
          @functools.wraps(func)
          def wrapper(*args, **kwargs):
              try:
                  return func(*args, **kwargs)
              except RuntimeError as exc:
                  logger.error("Runtime error %s: %s", operation, exc)
                  return f"Error {operation}: {exc}"
              except TransportServerError as exc:
                  code = getattr(exc, "code", "unknown")
                  logger.error(
                      "Monarch API HTTP %s error %s: %s", code, operation, exc,
                  )
                  return f"Error {operation}: Monarch API returned HTTP {code}: {exc}"
              except TransportQueryError as exc:
                  logger.error("Monarch API query error %s: %s", operation, exc)
                  return f"Error {operation}: API query failed: {exc}"
              except TransportError as exc:
                  logger.error(
                      "Monarch API connection error %s: %s", operation, exc,
                  )
                  return f"Error {operation}: connection error: {exc}"
              except Exception as exc:  # pylint: disable=broad-exception-caught
                  logger.error(
                      "Unexpected error %s: %s (%s)",
                      operation, exc, type(exc).__name__,
                  )
                  return f"Error {operation}: {exc}"
          return wrapper
      return decorator