#
tokens: 5124/50000 10/10 files
lines: off (toggle) GitHub
raw markdown copy
# 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?

![Claude roasting me](_photos/roast.png)
**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"])
```