update entrypoint.sh to improve script execution logic and README.md for clarity; add mcp and httpx to requirements

This commit is contained in:
Lukas Pierchalla
2026-09-20 12:07:52 +02:00
parent e7ecdfcd36
commit 9b4e351f79
4 changed files with 220 additions and 75 deletions
+76 -36
View File
@@ -10,14 +10,17 @@ import os
import sqlite3
import subprocess
import sys
import tempfile
import time
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Optional
import asyncio
import httpx
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
import requests
# ---------------------------------------------------------------------------
# Config
@@ -32,12 +35,8 @@ SC_SETTLED_ONLY = os.getenv("SC_SETTLED_ONLY", "true").lower() == "true"
SC_MAPPING_VERSION = int(os.getenv("SC_MAPPING_VERSION", "1"))
ASSET_MAP = json.loads(os.getenv("ASSET_MAP", "{}"))
# Scalable CLI Config: tmpfs-basiert, automatisch generiert
SC_CONFIG_DIR = Path(os.getenv("SC_CONFIG_DIR", tempfile.mkdtemp(prefix="sc-config-")))
SC_CONFIG_DIR.mkdir(parents=True, exist_ok=True)
(SC_CONFIG_DIR / "config.toml").write_text(
'[auth]\nsession_backend = "file"\nsigning_key_backend = "file"\n'
)
# SC_CONFIG_DIR wird vom Entrypoint gesetzt und als read-only Bind-Mount eingehängt
SC_CONFIG_DIR = Path(os.getenv("SC_CONFIG_DIR", "/tmp/sc-config"))
os.environ["SC_CONFIG_DIR"] = str(SC_CONFIG_DIR)
DATA_DIR = Path("/app/data")
@@ -260,38 +259,74 @@ def map_to_wealthfolio(tx: dict) -> list[WealthfolioActivity]:
# ---------------------------------------------------------------------------
class WealthfolioMCP:
def __init__(self, url, token):
def __init__(self, url: str, token: str):
self.url = url
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"Accept": "application/json",
})
self.token = token
def _rpc(self, method, params=None):
payload = {"jsonrpc": "2.0", "id": int(time.time() * 1000), "method": method}
if params: payload["params"] = params
resp = self.session.post(self.url, json=payload, timeout=30)
resp.raise_for_status()
return resp.json()
def _http_client(self) -> httpx.AsyncClient:
return httpx.AsyncClient(
headers={
"Authorization": f"Bearer {self.token}",
"Accept": "application/json, text/event-stream",
},
timeout=30.0,
)
def ping(self):
async def _connect_and_call(self, tool_name: str, arguments: dict):
async with self._http_client() as http_client:
async with streamable_http_client(
self.url,
http_client=http_client,
) as (read_stream, write_stream):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
return await session.call_tool(tool_name, arguments)
async def _list_tools(self) -> list[str]:
async with self._http_client() as http_client:
async with streamable_http_client(
self.url,
http_client=http_client,
) as (read_stream, write_stream):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
result = await session.list_tools()
return [tool.name for tool in result.tools]
def ping(self) -> bool:
try:
return self._rpc("ping").get("result") == "pong"
except Exception as e:
log.error("MCP ping failed: %s", e)
tools = asyncio.run(self._list_tools())
log.info("Available Wealthfolio MCP tools: %s", tools)
return True
except Exception as exc:
log.exception("MCP initialization failed: %s", exc)
return False
def record_activities(self, activities):
payload = [{
"accountId": a.account_id, "date": a.date, "symbol": a.symbol, "activityType": a.activity_type,
"quantity": a.quantity, "unitPrice": a.unit_price, "amount": a.amount,
"fee": a.fee, "tax": a.tax, "currency": a.currency,
"source": a.source, "externalId": a.external_id,
} for a in activities]
return self._rpc("tools/call", {"name": "record_activities", "arguments": {"activities": payload}})
payload = [
{
"accountId": activity.account_id,
"date": activity.date,
"symbol": activity.symbol,
"activityType": activity.activity_type,
"quantity": activity.quantity,
"unitPrice": activity.unit_price,
"amount": activity.amount,
"fee": activity.fee,
"tax": activity.tax,
"currency": activity.currency,
"source": activity.source,
"externalId": activity.external_id,
}
for activity in activities
]
return asyncio.run(
self._connect_and_call(
"record_activities",
{"activities": payload},
)
)
# ---------------------------------------------------------------------------
# Sync
@@ -310,9 +345,13 @@ def reconcile(mcp):
def sync_once(dry_run=False):
init_db()
mcp = WealthfolioMCP(WF_MCP_URL, WF_MCP_TOKEN)
if not mcp.ping():
raise RuntimeError("Wealthfolio MCP unreachable")
mcp = None
if not dry_run:
mcp = WealthfolioMCP(WF_MCP_URL, WF_MCP_TOKEN)
if not mcp.ping():
raise RuntimeError("Wealthfolio MCP unreachable")
from_time = datetime.now(timezone.utc) - timedelta(days=SC_OVERLAP_DAYS)
to_time = datetime.now(timezone.utc)
@@ -365,7 +404,8 @@ def sync_once(dry_run=False):
if not cursor or not items:
break
reconcile(mcp)
if not dry_run:
reconcile(mcp)
log.info("Sync complete: %s", stats)
return stats