Scripts Example Scripts

Scripts: Example Scripts

Runnable Python examples calling pp-mcp directly, without going through an AI assistant — e.g. to build your own reporting scripts or feed a dashboard. All examples use the official mcp Python SDK (the same package pp-mcp itself depends on) against a streamable-http server. See the note at the end for a stdio variant.

Prerequisites

pip install mcp httpx

A running pp-mcp instance reachable over HTTP (see Installation), e.g. http://localhost:8080/mcp. If MCP_AUTH_TOKEN is set, every example below needs the Authorization: Bearer <token> header shown in the connection helper.

Shared connection helper

All examples reuse this snippet — a small connect() async context manager wrapping the MCP handshake:

# pp_mcp_client.py
import json
from contextlib import asynccontextmanager

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

PP_MCP_URL = "http://localhost:8080/mcp"
MCP_AUTH_TOKEN = None  # set to your token string, or leave None if auth is disabled


@asynccontextmanager
async def connect():
    headers = {"Authorization": f"Bearer {MCP_AUTH_TOKEN}"} if MCP_AUTH_TOKEN else None
    async with streamablehttp_client(PP_MCP_URL, headers=headers) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            yield session


async def call(session: ClientSession, tool: str, **kwargs):
    """Calls a tool and returns its result as a plain Python object (dict/list/str/...)."""
    result = await session.call_tool(tool, kwargs)
    text = result.content[0].text  # pp-mcp tools return JSON-serialized text content
    data = json.loads(text)
    if isinstance(data, dict) and data.get("status") == "error":
        raise RuntimeError(f"{tool} failed: {data['message']}")
    return data

Every example below assumes this module is saved as pp_mcp_client.py next to the script.

1. Minimal client: connect and call ping

# 01_ping.py
import asyncio
from pp_mcp_client import connect, call


async def main():
    async with connect() as session:
        print(await call(session, "ping"))


asyncio.run(main())

Expected output:

pong

2. Print account balances for all accounts

# 02_account_balances.py
import asyncio
from pp_mcp_client import connect, call


async def main():
    async with connect() as session:
        accounts = await call(session, "list_accounts")
        for acc in accounts:
            balance = await call(session, "get_account_balance", account=acc["uuid"])
            print(f"{acc['name']:20} {balance['balance']:>12} {balance['currencyCode']}")


asyncio.run(main())

Expected output:

Broker                    1240.50 EUR
Savings                   8000.00 EUR

3. Export transactions of a date range to CSV

# 03_export_transactions_csv.py
import asyncio
import csv
import sys
from pp_mcp_client import connect, call


async def main(date_from: str, date_to: str, out_path: str):
    async with connect() as session:
        transactions = await call(session, "get_transactions", date_from=date_from, date_to=date_to)

    if not transactions:
        print("No transactions in this range.")
        return

    fieldnames = ["date", "type", "accountName", "portfolioName", "securityName", "amount", "currencyCode"]
    with open(out_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(transactions)
    print(f"Wrote {len(transactions)} transactions to {out_path}")


if __name__ == "__main__":
    asyncio.run(main("2025-01-01", "2025-12-31", "transactions_2025.csv"))

Expected output:

Wrote 214 transactions to transactions_2025.csv

4. Print current holdings with valuation

# 04_holdings.py
import asyncio
from pp_mcp_client import connect, call


async def main():
    async with connect() as session:
        holdings = await call(session, "get_holdings")

    for pos in holdings["positions"]:
        print(f"{pos['securityName']:30} {pos['shares']:>10} x {pos['price']:>10} = {pos['value']:>12} {pos['currencyCode']}")

    print("\nTotals:")
    for currency, total in holdings["totalsByCurrency"].items():
        print(f"  {total} {currency}")


asyncio.run(main())

Expected output:

iShares Core MSCI World            42.0000     84.20 =      3536.40 EUR
Apple Inc.                         10.0000    178.55 =      1785.50 USD
...

Totals:
  18420.30 EUR
  1785.50 USD

5. Plot portfolio value over time

# 05_plot_value_history.py
import asyncio
from datetime import date, timedelta

import matplotlib.pyplot as plt

from pp_mcp_client import connect, call


async def main():
    date_to = date.today().isoformat()
    date_from = (date.today() - timedelta(days=365)).isoformat()

    async with connect() as session:
        history = await call(
            session, "get_holdings_history",
            date_from=date_from, date_to=date_to, interval="monthly",
        )

    dates = [point["date"] for point in history]
    # Adjust "EUR" if your base currency differs — remember pp-mcp does NOT convert currencies.
    values = [float(point["totalsByCurrency"].get("EUR", 0)) for point in history]

    plt.plot(dates, values, marker="o")
    plt.xticks(rotation=45, ha="right")
    plt.ylabel("Value (EUR)")
    plt.title("Portfolio value over the last 12 months")
    plt.tight_layout()
    plt.savefig("portfolio_value.png")
    print("Saved chart to portfolio_value.png")


asyncio.run(main())

Requires pip install matplotlib in addition to the base prerequisites.

6. Multi-source: aggregate account balances across all sources

For a pp-mcp instance configured with PP_PORTFOLIOS_CONFIG (see Installation):

# 06_multi_source_balances.py
import asyncio
from pp_mcp_client import connect, call


async def main():
    async with connect() as session:
        sources = await call(session, "list_data_sources")

        for src in sources:
            source_id = src["id"]
            print(f"\n== {src['label']} ({source_id}) ==")
            accounts = await call(session, "list_accounts", source=source_id)
            for acc in accounts:
                balance = await call(session, "get_account_balance", account=acc["uuid"], source=source_id)
                print(f"  {acc['name']:20} {balance['balance']:>12} {balance['currencyCode']}")


asyncio.run(main())

Expected output:

== Example1 (example1) ==
  Broker                    1240.50 EUR
  Savings                   8000.00 EUR

== Example2 (example2) ==
  Broker                    3120.00 EUR

Using stdio instead of streamable-http

If pp-mcp isn't running as a resident HTTP server (see Configuring AI Tools), replace the connection helper's streamablehttp_client(...) block with stdio_client, which starts pp-mcp itself as a subprocess:

from mcp import StdioServerParameters
from mcp.client.stdio import stdio_client

server_params = StdioServerParameters(
    command="python3",
    args=["-m", "src.main"],
    env={"PYTHONPATH": "/path/to/pp-mcp", "MCP_TRANSPORT": "stdio", "PP_FILE_PATH": "/path/to/file.portfolio"},
    cwd="/path/to/pp-mcp",
)

async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()
        # ... same call() helper works unchanged from here

No bearer token needed here — there's no HTTP layer for stdio.

---

← Function Reference · Home