get_upcoming_bills

Instructions

Implementation Reference

• tuskledger_mcp/server.py:155-171 (schema)

  Tool definition (schema) for 'get_upcoming_bills': registers the MCP tool with name, description, and input schema (optional 'days' integer parameter). Included in the TOOLS list that is returned by list_tools().

  Tool(
      name="get_upcoming_bills",
      description=(
          "Forward 30-day calendar of expected bills + paychecks with a "
          "running balance. Returns each event's date, amount, source "
          "(merchant or paycheck), and the projected account balance "
          "after that event. Useful for 'is my account going to dip "
          "before payday?'."
      ),
      inputSchema={
          "type": "object",
          "properties": {
              "days": {"type": "integer", "description": "How many days forward to look (default 30)."},
          },
          "additionalProperties": False,
      },
  ),

• tuskledger_mcp/server.py:293-294 (handler)

  Tool handler (dispatcher) for 'get_upcoming_bills' in _dispatch(): calls client.upcoming_bills() passing the 'days' argument from the caller (if provided).

  if name == "get_upcoming_bills":
      return client.upcoming_bills(**{k: v for k, v in a.items() if v not in (None, "")})

• tuskledger_mcp/client.py:147-149 (helper)

  HTTP client helper method 'upcoming_bills' that sends a GET request to '/api/bills/upcoming' with optional query params (e.g., 'days'), returning a list of dicts.

  # bills
  def upcoming_bills(self, **params) -> list[dict]:
      return self._request("GET", "/api/bills/upcoming", params=params)

• tuskledger_mcp/server.py:329-366 (registration)

  Server wiring in build_server() where list_tools() and call_tool() are registered as MCP handlers; the latter calls _dispatch() which routes to the get_upcoming_bills handler.

  @server.list_tools()
  async def list_tools() -> list[Tool]:
      return TOOLS
  
  @server.call_tool()
  async def call_tool(name: str, arguments: dict) -> list[TextContent]:
      log.info("tool call: %s args=%s", name, list((arguments or {}).keys()))
      try:
          # The HTTP calls are blocking; offload to a thread so we don't
          # stall the event loop. Backend is on localhost so latency is
          # tiny but doing this correctly keeps the door open for
          # async-aware tools later.
          payload = await asyncio.to_thread(_dispatch, name, arguments, cli)
          return [TextContent(type="text", text=_format_result(payload))]
      except TuskLedgerError as e:
          # Surface the error in a way the assistant can show the user.
          err_payload = {
              "error": True,
              "message": str(e),
              "status": e.status,
              "body": e.body,
              "hint": (
                  "If the backend is unreachable, run `./start.sh` from the "
                  "repo root. If the endpoint returned 404 or 500, run "
                  "`./tuskledger doctor --json` for a structured health check."
              ),
          }
          return [TextContent(type="text", text=_format_result(err_payload))]
      except Exception as e:  # pylint: disable=broad-except
          log.exception("tool %s crashed", name)
          err_payload = {
              "error": True,
              "message": f"Unexpected error in {name!r}: {type(e).__name__}: {e}",
              "hint": "Likely a bug in tuskledger-mcp; please file an issue.",
          }
          return [TextContent(type="text", text=_format_result(err_payload))]
  
  return server