monarch-mcp

Instructions

Implementation Reference

• src/monarch_mcp/server.py:1132-1148 (handler)

  The main handler function for the 'get_account_history' tool. It's decorated with @mcp.tool() to register it as an MCP tool, and uses @_handle_mcp_errors for error handling. The function takes an account_id parameter, creates an async client, calls client.get_account_history(), and returns the result as JSON.

  @mcp.tool()
  @_handle_mcp_errors("getting account history")
  def get_account_history(account_id: str) -> str:
      """
      Get historical balance snapshots for an account.
  
      Args:
          account_id: The ID of the account
      """
  
      async def _get_account_history():
          client = await get_monarch_client()
          return await client.get_account_history(account_id)
  
      history = run_async(_get_account_history())
  
      return json.dumps(history, indent=2, default=str)

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

  The @mcp.tool() decorator registers the get_account_history function as an MCP tool with the FastMCP framework. The tool name is derived from the function name.

  @mcp.tool()

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

  Error handling decorator that wraps the tool handler. It catches specific exceptions (RuntimeError, TransportServerError, TransportQueryError, TransportError) and returns user-friendly error messages, ensuring the MCP tool never crashes.

  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

• src/monarch_mcp/server.py:53-77 (helper)

  Helper function that runs async coroutines in a thread pool with proper event loop handling. Also handles authentication errors by clearing expired tokens and triggering re-authentication flow.

  def run_async(coro):
      """Run async function in a new thread with its own event loop.
  
      If the coroutine raises an authentication error (expired token,
      invalid credentials), the stale token is cleared from the keyring,
      the browser-based auth flow is re-triggered, and a RuntimeError is
      raised so the calling tool can inform the user.
  
      Only catches the two exception types that ``is_auth_error`` can
      recognise; everything else propagates unchanged to the caller.
      """
      with ThreadPoolExecutor() as executor:
          future = executor.submit(_run_sync, coro)
          try:
              return future.result()
          except (TransportServerError, LoginFailedException) as exc:
              if is_auth_error(exc):
                  logger.warning("Token appears expired — clearing and triggering re-auth")
                  secure_session.delete_token()
                  trigger_auth_flow()
                  raise RuntimeError(
                      "Your session has expired. A login page has been opened in "
                      "your browser — please sign in and try again."
                  ) from exc
              raise

• src/monarch_mcp/server.py:123-157 (helper)

  Helper function that retrieves or creates a MonarchMoney client instance. It checks secure session storage first, then falls back to environment credentials, and finally triggers browser-based authentication if needed.

  async def get_monarch_client() -> MonarchMoney:
      """Get or create MonarchMoney client instance using secure session storage."""
      # Try to get authenticated client from secure session
      client = secure_session.get_authenticated_client()
  
      if client is not None:
          logger.info("Using authenticated client from secure keyring storage")
          return client
  
      # If no secure session, try environment credentials
      email = os.getenv("MONARCH_EMAIL")
      password = os.getenv("MONARCH_PASSWORD")
  
      if email and password:
          try:
              client = MonarchMoney()
              await client.login(email, password)
              logger.info(
                  "Successfully logged into Monarch Money with environment credentials"
              )
  
              # Save the session securely
              secure_session.save_authenticated_session(client)
  
              return client
          except Exception as e:
              logger.error("Failed to login to Monarch Money: %s", e)
              raise
  
      # No credentials anywhere — open browser login and tell the user
      trigger_auth_flow()
      raise RuntimeError(
          "Authentication needed! A login page has been opened in your "
          "browser — please sign in and try again."
      )