monarch-mcp

Instructions

Implementation Reference

• src/monarch_mcp/server.py:448-464 (handler)

  The main handler function for the 'get_account_holdings' MCP tool. It's decorated with @mcp.tool() for registration and @_handle_mcp_errors for error handling. The function takes an account_id parameter, creates an async helper that gets the MonarchMoney client and calls client.get_account_holdings(account_id), then returns the JSON-serialized result.

  @mcp.tool()
  @_handle_mcp_errors("getting account holdings")
  def get_account_holdings(account_id: str) -> str:
      """
      Get investment holdings for a specific account.
  
      Args:
          account_id: The ID of the investment account
      """
  
      async def _get_holdings():
          client = await get_monarch_client()
          return await client.get_account_holdings(account_id)
  
      holdings = run_async(_get_holdings())
  
      return json.dumps(holdings, indent=2, default=str)

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

  The _handle_mcp_errors decorator that wraps the get_account_holdings handler to provide comprehensive error handling. It catches RuntimeError, TransportServerError, TransportQueryError, TransportError, and generic Exception, 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

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

  The get_monarch_client async helper function that creates and returns a MonarchMoney client instance. It first tries to get an authenticated client from secure session storage, then falls back to environment credentials (MONARCH_EMAIL, MONARCH_PASSWORD).

  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

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

  The @mcp.tool() decorator registers the get_account_holdings function as an MCP tool with the FastMCP server. FastMCP automatically generates the input schema from the function signature (account_id: str).

  @mcp.tool()