# Directory Structure ``` ├── _photos │ └── roast.png ├── .env.example ├── .gitignore ├── .python-version ├── LICENSE ├── main.py ├── pyproject.toml ├── README.md ├── requirements.txt ├── test.py ├── up_mcp.py ├── uv.lock └── weather.py ``` # Files -------------------------------------------------------------------------------- /.python-version: -------------------------------------------------------------------------------- ``` 3.13 ``` -------------------------------------------------------------------------------- /.env.example: -------------------------------------------------------------------------------- ``` UP_TOKEN="up:yeah:YOUR_UP_TOKEN_HERE" ``` -------------------------------------------------------------------------------- /.gitignore: -------------------------------------------------------------------------------- ``` __pycache__/ .DS_Store .venv/ .mypy_cache/ .pytest_cache/ .ruff_cache/ .vscode/ .env ``` -------------------------------------------------------------------------------- /README.md: -------------------------------------------------------------------------------- ```markdown # Up Bank MCP Server MCP wrapper of the Python Up Bank API Wrapper [up-bank-api](https://github.com/jcwillox/up-bank-api) (credit to [@jcwillox](https://github.com/jcwillox)) (it's wrappers all the way down) Made for Claude, by Claude (lol) This is a Model Context Protocol (MCP) server that provides tools for interacting with the Up Bank API. It allows you to manage accounts, transactions, categories, tags, and webhooks through MCP-enabled clients like Claude for Desktop. Feel free to contribute if you want to better optimise it for LLM, etc. However it's fine as is. I just wanted to learn how to make an MCP server. Now Claude can roast me for my transaction history, what else is it good for?  **Figure: Too much takeout I guess** ## Prerequisites - Python 3.10 or higher - Up Bank API token - `uv` package manager ## Installation 1. First, install `uv` by running: ```bash curl -LsSf https://astral.sh/uv/install.sh | sh ``` Make sure to restart your terminal after installing `uv`. 2. Clone this repository and navigate to it: ```bash git clone <repository-url> cd up-mcp ``` 3. Create and activate a virtual environment: ```bash uv venv source .venv/bin/activate # On Windows, use: .venv\Scripts\activate ``` 4. Install dependencies: ```bash uv pip install -r requirements.txt ``` ### Using with Claude for Desktop or other MCP-enabled clients 1. Open your Claude (or other MCP-enabled) configuration file: - macOS/Linux: `~/Library/Application Support/Claude/claude_desktop_config.json` - Windows: `%AppData%\Claude\claude_desktop_config.json` 2. Add the server configuration: ```json { "mcpServers": { "up-mcp": { "command": "/ABSOLUTE/PATH/TO/uv", "args": [ "--directory", "/ABSOLUTE/PATH/TO/up-mcp", "run", "up_mcp.py" ], "env": { "UP_TOKEN": "up:yeah:your-token-here" } } } } ``` Replace `/ABSOLUTE/PATH/TO/uv` with the absolute path to your `uv` executable. Replace `/ABSOLUTE/PATH/TO/up-mcp` with the absolute path to your project directory. Replace `up:yeah:your-token-here` with your Up Bank API token. Get your Up Bank API token from the [Up Bank website (https://api.up.com.au/)](https://api.up.com.au/). 3. Restart Claude for Desktop. ## Available Tools The server provides the following tools: - Account management (get accounts, get specific account) - Transaction management (get transactions, get specific transaction) - Category management (get categories, categorize transactions) - Tag management (get tags, add/remove tags from transactions) - Webhook management (create, delete, ping webhooks) ## Testing You can test the server using the included `test.py` script: ```bash python test.py ``` This will run through basic functionality tests including account retrieval and transaction listing. ## Troubleshooting If you encounter issues: 1. Verify your UP_TOKEN environment variable is set correctly 2. Check that all dependencies are installed correctly 3. Ensure you're using Python 3.10 or higher 4. Check Claude's logs for MCP-related issues ``` -------------------------------------------------------------------------------- /requirements.txt: -------------------------------------------------------------------------------- ``` mcp[cli] httpx up-bank-api pytest-asyncio ``` -------------------------------------------------------------------------------- /main.py: -------------------------------------------------------------------------------- ```python import weather def main(): weather.mcp.run(transport='stdio') if __name__ == "__main__": main() ``` -------------------------------------------------------------------------------- /pyproject.toml: -------------------------------------------------------------------------------- ```toml [project] name = "up-mcp" version = "0.1.0" description = "Add your description here" readme = "README.md" requires-python = ">=3.13" dependencies = [ "httpx>=0.28.1", "mcp[cli]>=1.6.0", ] ``` -------------------------------------------------------------------------------- /weather.py: -------------------------------------------------------------------------------- ```python from typing import Any import httpx from mcp.server.fastmcp import FastMCP # Initialize FastMCP server mcp = FastMCP("weather") # Constants NWS_API_BASE = "https://api.weather.gov" USER_AGENT = "weather-app/1.0" async def make_nws_request(url: str) -> dict[str, Any] | None: """Make a request to the NWS API with proper error handling.""" headers = { "User-Agent": USER_AGENT, "Accept": "application/geo+json" } async with httpx.AsyncClient() as client: try: response = await client.get(url, headers=headers, timeout=30.0) response.raise_for_status() return response.json() except Exception: return None def format_alert(feature: dict) -> str: """Format an alert feature into a readable string.""" props = feature["properties"] return f""" Event: {props.get('event', 'Unknown')} Area: {props.get('areaDesc', 'Unknown')} Severity: {props.get('severity', 'Unknown')} Description: {props.get('description', 'No description available')} Instructions: {props.get('instruction', 'No specific instructions provided')} """ @mcp.tool() async def get_alerts(state: str) -> str: """Get weather alerts for a US state. Args: state: Two-letter US state code (e.g. CA, NY) """ url = f"{NWS_API_BASE}/alerts/active/area/{state}" data = await make_nws_request(url) if not data or "features" not in data: return "Unable to fetch alerts or no alerts found." if not data["features"]: return "No active alerts for this state." alerts = [format_alert(feature) for feature in data["features"]] return "\n---\n".join(alerts) @mcp.tool() async def get_forecast(latitude: float, longitude: float) -> str: """Get weather forecast for a location. Args: latitude: Latitude of the location longitude: Longitude of the location """ # First get the forecast grid endpoint points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}" points_data = await make_nws_request(points_url) if not points_data: return "Unable to fetch forecast data for this location." # Get the forecast URL from the points response forecast_url = points_data["properties"]["forecast"] forecast_data = await make_nws_request(forecast_url) if not forecast_data: return "Unable to fetch detailed forecast." # Format the periods into a readable forecast periods = forecast_data["properties"]["periods"] forecasts = [] for period in periods[:5]: # Only show next 5 periods forecast = f""" {period['name']}: Temperature: {period['temperature']}°{period['temperatureUnit']} Wind: {period['windSpeed']} {period['windDirection']} Forecast: {period['detailedForecast']} """ forecasts.append(forecast) return "\n---\n".join(forecasts) if __name__ == "__main__": # Initialize and run the server mcp.run(transport='stdio') ``` -------------------------------------------------------------------------------- /up_mcp.py: -------------------------------------------------------------------------------- ```python from upbankapi import AsyncClient, NotAuthorizedException from typing import Any, Optional, Union from datetime import datetime, timedelta from mcp.server.fastmcp import FastMCP import os # Initialize FastMCP server mcp = FastMCP("up-mcp") UP_TOKEN = os.getenv("UP_TOKEN") @mcp.tool() async def get_user_id() -> str: """Get the user ID for the UP API. """ async with AsyncClient(token=UP_TOKEN) as client: try: user_id = await client.ping() return f"Authorized: {user_id}" except NotAuthorizedException: return "The token is invalid" @mcp.tool() async def get_accounts() -> list[dict[str, Any]]: """Get all accounts for the user. """ async with AsyncClient(token=UP_TOKEN) as client: accounts = await client.accounts() account_list = [] async for account in accounts: account_list.append({ "id": account.id, "name": account.name, "balance": account.balance }) return account_list @mcp.tool() async def get_account(id: str) -> str: """Get an account for the user. Args: id: The ID of the account to get. """ async with AsyncClient(token=UP_TOKEN) as client: account = await client.account(id) return str(account) @mcp.tool() async def get_categories(parent_id: Optional[str] = None) -> list[dict[str, Any]]: """Get all categories or categories under a specific parent. Args: parent_id: Optional ID of the parent category to filter by. """ async with AsyncClient(token=UP_TOKEN) as client: categories = await client.categories(parent=parent_id) return [{"id": cat.id, "name": cat.name} for cat in categories] @mcp.tool() async def get_category(category_id: str) -> str: """Get a specific category by ID. Args: category_id: The ID of the category to get. """ async with AsyncClient(token=UP_TOKEN) as client: category = await client.category(category_id) return str(category) @mcp.tool() async def categorize_transaction(transaction_id: str, category_id: Optional[str]) -> bool: """Categorize a transaction. Args: transaction_id: The ID of the transaction to categorize. category_id: The category ID to assign, or None to remove categorization. """ async with AsyncClient(token=UP_TOKEN) as client: return await client.categorize(transaction_id, category_id) @mcp.tool() async def get_tags() -> list[dict[str, Any]]: """Get all tags for the user.""" async with AsyncClient(token=UP_TOKEN) as client: tags = await client.tags() tag_list = [] async for tag in tags: tag_list.append({"id": tag.id, "name": tag.name}) return tag_list @mcp.tool() async def add_transaction_tags(transaction_id: str, tags: list[str]) -> bool: """Add tags to a transaction. Args: transaction_id: The ID of the transaction. tags: List of tag IDs to add. """ async with AsyncClient(token=UP_TOKEN) as client: return await client.add_tags(transaction_id, *tags) @mcp.tool() async def remove_transaction_tags(transaction_id: str, tags: list[str]) -> bool: """Remove tags from a transaction. Args: transaction_id: The ID of the transaction. tags: List of tag IDs to remove. """ async with AsyncClient(token=UP_TOKEN) as client: return await client.remove_tags(transaction_id, *tags) @mcp.tool() async def get_transaction(transaction_id: str) -> str: """Get a specific transaction by ID. Args: transaction_id: The ID of the transaction to get. """ async with AsyncClient(token=UP_TOKEN) as client: transaction = await client.transaction(transaction_id) return str(transaction) @mcp.tool() async def get_transactions( account_id: Optional[str] = None, status: Optional[str] = None, since: Optional[datetime] = datetime.now() - timedelta(days=7), until: Optional[datetime] = None, category_id: Optional[str] = None, tag_id: Optional[str] = None, verbose: bool = False ) -> list[dict[str, Any]]: """Get transactions with optional filters. Args: account_id: Optional account ID to filter by. status: Optional transaction status to filter by. since: Optional start datetime (defaults to 7 days ago - longer may take longer to load) until: Optional end datetime. category_id: Optional category ID to filter by. tag_id: Optional tag ID to filter by. """ async with AsyncClient(token=UP_TOKEN) as client: transactions = await client.transactions( account=account_id, status=status, since=since, until=until, category=category_id, tag=tag_id ) transaction_list = [] if verbose: async for tx in transactions: transaction_list.append({ "id": tx.id, "description": tx.description, "amount": tx.amount, "status": tx.status, "created_at": tx.created_at }) else: async for tx in transactions: transaction_list.append({ "description": tx.description, "amount": tx.amount, }) return transaction_list @mcp.tool() async def get_webhooks() -> list[dict[str, Any]]: """Get all webhooks for the user.""" async with AsyncClient(token=UP_TOKEN) as client: webhooks = await client.webhooks() webhook_list = [] async for webhook in webhooks: webhook_list.append({ "id": webhook.id, "url": webhook.url, "description": webhook.description }) return webhook_list @mcp.tool() async def create_webhook(url: str, description: Optional[str] = None) -> dict[str, Any]: """Create a new webhook. Args: url: The URL that this webhook should post events to. """ async with AsyncClient(token=UP_TOKEN) as client: webhook = await client.webhook.create(url, description) return { "id": webhook.id, "url": webhook.url, "description": webhook.description, "secret_key": webhook.secret_key, "created_at": webhook.created_at } @mcp.tool() async def delete_webhook(webhook_id: str) -> bool: """Delete a webhook. Args: webhook_id: The ID of the webhook to delete. """ async with AsyncClient(token=UP_TOKEN) as client: return await client.webhook.delete(webhook_id) @mcp.tool() async def ping_webhook(webhook_id: str) -> str: """Ping a webhook. Args: webhook_id: The ID of the webhook to ping. """ async with AsyncClient(token=UP_TOKEN) as client: event = await client.webhook.ping(webhook_id) return str(event) if __name__ == "__main__": mcp.run(transport='stdio') ``` -------------------------------------------------------------------------------- /test.py: -------------------------------------------------------------------------------- ```python import pytest import asyncio from datetime import datetime, timedelta import os from typing import AsyncGenerator import up_mcp from upbankapi import AsyncClient, NotAuthorizedException # Fixture to ensure UP_TOKEN is set @pytest.fixture(scope="session", autouse=True) def check_token(): token = os.getenv("UP_TOKEN") if not token: pytest.skip("UP_TOKEN environment variable not set") return token # Fixture for async tests @pytest.fixture(scope="session") def event_loop(): loop = asyncio.get_event_loop() yield loop loop.close() @pytest.fixture(scope="session") async def client(check_token) -> AsyncGenerator[AsyncClient, None]: async with AsyncClient(token=check_token) as client: yield client @pytest.mark.asyncio async def test_get_user_id(): """Test that we can authenticate and get a user ID.""" result = await up_mcp.get_user_id() assert result is not None assert isinstance(result, str) # Extract the actual user ID from the result # The format is "Authorized: {user_id}" assert result.startswith("Authorized: ") user_id = result.split("Authorized: ")[1] # Check that the user_id is a valid UUID format assert isinstance(user_id, str) # Optional: validate UUID format if needed import uuid try: uuid_obj = uuid.UUID(user_id) assert str(uuid_obj) == user_id # Ensures it's a valid UUID string except ValueError: pytest.fail(f"User ID '{user_id}' is not a valid UUID") @pytest.mark.asyncio async def test_get_accounts(): """Test retrieving all accounts.""" accounts = await up_mcp.get_accounts() assert accounts is not None assert isinstance(accounts, list) assert len(accounts) > 0 # Check account structure based on actual response account = accounts[0] assert "id" in account assert "name" in account assert "balance" in account assert isinstance(account["id"], str) assert isinstance(account["name"], str) assert isinstance(account["balance"], (int, float)) @pytest.mark.asyncio async def test_get_specific_account(): """Test retrieving a specific account by ID.""" # First get all accounts accounts = await up_mcp.get_accounts() account_id = accounts[0]["id"] # Then get specific account account = await up_mcp.get_account(account_id) assert account is not None # Check the structure of the response # The actual response is formatted like: <Account 'Spending' (TRANSACTIONAL): 9.29 AUD> assert isinstance(account, str) assert account.startswith("<Account ") assert account.endswith(">") # Check that the account name from the first response is in the specific account response account_name = accounts[0]["name"] assert account_name in account @pytest.mark.asyncio async def test_get_transactions(): """Test retrieving transactions with various filters.""" # Test with date filter since_date = datetime.now() - timedelta(days=7) transactions = await up_mcp.get_transactions(since=since_date) assert transactions is not None assert isinstance(transactions, list) if len(transactions) > 0: # Check transaction structure based on actual response transaction = transactions[0] assert isinstance(transaction, dict) assert "description" in transaction assert "amount" in transaction assert isinstance(transaction["description"], str) assert isinstance(transaction["amount"], (int, float)) # Test with verbose parameter verbose_transactions = await up_mcp.get_transactions(since=since_date, verbose=True) if len(verbose_transactions) > 0: verbose_tx = verbose_transactions[0] assert "id" in verbose_tx assert "status" in verbose_tx assert "created_at" in verbose_tx @pytest.mark.asyncio async def test_get_categories(): """Test retrieving categories.""" categories = await up_mcp.get_categories() assert categories is not None assert isinstance(categories, list) if len(categories) > 0: # Check category structure based on actual response category = categories[0] assert "id" in category assert "name" in category assert isinstance(category["id"], str) assert isinstance(category["name"], str) @pytest.mark.asyncio async def test_invalid_account_id(): """Test error handling for invalid account ID.""" with pytest.raises(Exception): # Replace with specific exception when implemented await up_mcp.get_account("invalid-account-id") @pytest.mark.asyncio async def test_get_transaction(): """Test retrieving a specific transaction.""" # First get all transactions with verbose to get IDs transactions = await up_mcp.get_transactions(verbose=True) if len(transactions) > 0: transaction_id = transactions[0]["id"] # Get the specific transaction transaction = await up_mcp.get_transaction(transaction_id) assert transaction is not None # Check that the response is formatted properly # Actual response format is like: <Transaction HELD: -34.0 AUD [Claude Ai]> assert isinstance(transaction, str) assert transaction.startswith("<Transaction ") assert transaction.endswith(">") # Verify the transaction amount and description are present in the response tx_amount = str(abs(transactions[0]["amount"])) tx_desc = transactions[0]["description"] # The specific format might vary, but amount and description should be in the string assert tx_amount in transaction assert tx_desc in transaction @pytest.mark.asyncio async def test_categorize_transaction(): """Test categorizing a transaction.""" # First get all transactions with verbose to get IDs transactions = await up_mcp.get_transactions(verbose=True) if len(transactions) > 0: # Get categories to find a valid category ID categories = await up_mcp.get_categories() if len(categories) > 0: transaction_id = transactions[0]["id"] category_id = categories[0]["id"] # Attempt to categorize a transaction result = await up_mcp.categorize_transaction( transaction_id=transaction_id, category_id=category_id ) # Verify the result assert result is not None assert isinstance(result, bool) assert result == True @pytest.mark.asyncio async def test_webhooks(): """Test webhook operations.""" # Get existing webhooks webhooks = await up_mcp.get_webhooks() # Should at least return an empty list without error assert isinstance(webhooks, list) # Test creating a webhook test_url = "https://hello.requestcatcher.com/test" result = await up_mcp.create_webhook(url=test_url) # Verify webhook creation result assert result is not None assert isinstance(result, dict) assert "id" in result webhook_id = result["id"] # Test pinging the webhook ping_result = await up_mcp.ping_webhook(webhook_id=webhook_id) assert ping_result is not None # Test deleting the webhook delete_result = await up_mcp.delete_webhook(webhook_id=webhook_id) assert delete_result is not None # Verify webhook was deleted by checking the list again updated_webhooks = await up_mcp.get_webhooks() assert all(webhook["id"] != webhook_id for webhook in updated_webhooks) if __name__ == "__main__": pytest.main([__file__, "-v"]) ```