Vladimir Buzalka
730 lines
28 KiB
Python
730 lines
28 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
MCP server pro Medevio API — FastMCP
|
|
Spustit: python mcp_medevio.py
|
|
|
|
Nástroje:
|
|
get_agenda — agenda kalendáře (jeden den nebo rozsah)
|
|
zapis_poznamku — vytvoř poznámku lékaře v kalendáři
|
|
smaz_rezervaci — zruš jednorazovou rezervaci/poznámku
|
|
get_pozadavky — seznam požadavků (aktivní nebo vyřízené)
|
|
get_pozadavek — detail jednoho požadavku
|
|
get_poznamky — interní klinické poznámky k požadavku
|
|
uloz_poznamku — vytvoř nebo aktualizuj interní poznámku k požadavku
|
|
hledej_pacienta — vyhledej pacienta podle jména / rodného čísla
|
|
get_pacient — detail pacienta podle UUID
|
|
|
|
TODO:
|
|
- smaz_rezervaci neumí mazat opakované (recurring) poznámky — potřeba
|
|
zachytit GraphQL mutaci z webu Medevio při smazání recurring reservation
|
|
a přidat podporu (smazat celou sérii / jeden výskyt).
|
|
"""
|
|
|
|
import sys
|
|
import json
|
|
import traceback
|
|
from pathlib import Path
|
|
from datetime import datetime, date, time as dtime, timedelta
|
|
from typing import Optional
|
|
|
|
import requests
|
|
from dateutil import parser as dtparser, tz
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
# ── Všechny logy na stderr (stdout = JSON-RPC) ──────────────────────────────
|
|
def log(msg: str):
|
|
print(msg, file=sys.stderr, flush=True)
|
|
|
|
# ── Konstanty ────────────────────────────────────────────────────────────────
|
|
GRAPHQL_URL = "https://api.medevio.cz/graphql"
|
|
CLINIC_SLUG = "mudr-buzalkova"
|
|
PRAGUE_TZ = tz.gettz("Europe/Prague")
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
TOKEN_PATH = BASE_DIR / "Medevio" / "token.txt"
|
|
STORAGE_PATH = BASE_DIR / "Medevio" / "medevio_storage.json" # Playwright session cookies
|
|
|
|
CALENDARS = {
|
|
"vlado": "b6555c7e-4e95-4657-b441-87c2c9a7b2ca",
|
|
"manzelka": "144c4e12-347c-49ca-9ec0-8ca965a4470d",
|
|
}
|
|
|
|
_BASE_HEADERS = {
|
|
"content-type": "application/json",
|
|
"origin": "https://my.medevio.cz",
|
|
"referer": "https://my.medevio.cz/",
|
|
}
|
|
|
|
# ── Auth / token refresh ──────────────────────────────────────────────────────
|
|
|
|
def _load_token() -> str:
|
|
t = TOKEN_PATH.read_text(encoding="utf-8").strip()
|
|
if t.startswith("Bearer "):
|
|
t = t.split(" ", 1)[1]
|
|
return t
|
|
|
|
def _save_token(token: str) -> None:
|
|
TOKEN_PATH.write_text(token, encoding="utf-8")
|
|
log(f"✓ Token obnoven a uložen do {TOKEN_PATH}")
|
|
|
|
def _load_storage_cookies() -> dict:
|
|
"""Načte session cookies z Playwright storage.json (gateway-access-token aj.)."""
|
|
if not STORAGE_PATH.exists():
|
|
raise FileNotFoundError(f"Nenalezen soubor session: {STORAGE_PATH}")
|
|
state = json.loads(STORAGE_PATH.read_text(encoding="utf-8"))
|
|
return {c["name"]: c["value"] for c in state.get("cookies", [])}
|
|
|
|
def _refresh_token_via_cookies() -> str:
|
|
"""
|
|
Obnoví Bearer token pomocí session cookies z medevio_storage.json.
|
|
Volá AccessToken_AuthSelf s cookies → dostane nový JWT → uloží do token.txt.
|
|
"""
|
|
log("⟳ Obnova Bearer tokenu pomocí session cookies...")
|
|
cookies = _load_storage_cookies()
|
|
payload = {
|
|
"operationName": "AccessToken_AuthSelf",
|
|
"query": "query AccessToken_AuthSelf { authSelf { token } }",
|
|
"variables": {},
|
|
}
|
|
r = requests.post(
|
|
GRAPHQL_URL,
|
|
headers=_BASE_HEADERS,
|
|
cookies=cookies,
|
|
data=json.dumps(payload),
|
|
timeout=20,
|
|
)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
if "errors" in data:
|
|
raise RuntimeError(f"AccessToken_AuthSelf selhalo: {data['errors']}")
|
|
token = data["data"]["authSelf"]["token"]
|
|
if not token:
|
|
raise RuntimeError("AccessToken_AuthSelf vrátilo prázdný token.")
|
|
_save_token(token)
|
|
return token
|
|
|
|
def _refresh_token_via_playwright() -> str:
|
|
"""
|
|
Záložní metoda: spustí headless Playwright, načte storage.json,
|
|
přejde na Medevio a zachytí Bearer token z prvního GraphQL requestu.
|
|
Použije se jen pokud i session cookies expirují.
|
|
"""
|
|
log("⟳ Záložní refresh přes Playwright (headless)...")
|
|
from playwright.sync_api import sync_playwright
|
|
import threading
|
|
|
|
captured = []
|
|
done = threading.Event()
|
|
|
|
def on_request(req):
|
|
if "graphql" in req.url and not done.is_set():
|
|
auth = req.headers.get("authorization", "")
|
|
if auth.startswith("Bearer "):
|
|
token = auth.split(" ", 1)[1]
|
|
captured.append(token)
|
|
done.set()
|
|
|
|
with sync_playwright() as pw:
|
|
browser = pw.chromium.launch(headless=True)
|
|
ctx = browser.new_context(storage_state=str(STORAGE_PATH))
|
|
page = ctx.new_page()
|
|
page.on("request", on_request)
|
|
page.goto(
|
|
"https://my.medevio.cz/mudr-buzalkova/klinika/kalendar/agenda-dne/",
|
|
wait_until="networkidle",
|
|
timeout=30_000,
|
|
)
|
|
# Počkáme max 10 s na zachycení tokenu
|
|
done.wait(timeout=10)
|
|
browser.close()
|
|
|
|
if not captured:
|
|
raise RuntimeError(
|
|
"Playwright nenachytil Bearer token. "
|
|
"Pravděpodobně expirovala i Playwright session — přihlaš se ručně na "
|
|
"https://my.medevio.cz a zkopíruj nový Bearer token do Medevio/token.txt."
|
|
)
|
|
_save_token(captured[0])
|
|
return captured[0]
|
|
|
|
def _refresh_token() -> str:
|
|
"""Zkusí obnovit token: nejdřív přes cookies, při selhání přes Playwright."""
|
|
try:
|
|
return _refresh_token_via_cookies()
|
|
except Exception as e:
|
|
log(f"Cookie refresh selhal ({e}), zkouším Playwright...")
|
|
return _refresh_token_via_playwright()
|
|
|
|
def _headers(token: Optional[str] = None) -> dict:
|
|
t = token or _load_token()
|
|
return {**_BASE_HEADERS, "authorization": f"Bearer {t}"}
|
|
|
|
def _gql(operation: str, query: str, variables: dict) -> dict:
|
|
"""Zavolá GraphQL. Při 401/403 nebo auth chybě automaticky obnoví token a zkusí znovu."""
|
|
payload = {"operationName": operation, "query": query, "variables": variables}
|
|
|
|
def _do(hdrs: dict) -> requests.Response:
|
|
return requests.post(GRAPHQL_URL, headers=hdrs, data=json.dumps(payload), timeout=30)
|
|
|
|
r = _do(_headers())
|
|
|
|
# Automatický refresh při vypršeném tokenu
|
|
if r.status_code in (401, 403):
|
|
log(f"⚠ HTTP {r.status_code} — obnova tokenu...")
|
|
new_token = _refresh_token()
|
|
r = _do(_headers(new_token))
|
|
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
|
|
# GraphQL může vrátit auth chybu s HTTP 200
|
|
if "errors" in data:
|
|
errs = data["errors"]
|
|
is_auth = any(
|
|
"auth" in str(e).lower() or "unauthenticated" in str(e).lower()
|
|
or "unauthorized" in str(e).lower()
|
|
for e in errs
|
|
)
|
|
if is_auth:
|
|
log("⚠ GraphQL auth chyba — obnova tokenu...")
|
|
new_token = _refresh_token()
|
|
r = _do(_headers(new_token))
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
if "errors" in data:
|
|
raise RuntimeError(f"GraphQL error [{operation}]: {data['errors']}")
|
|
|
|
return data["data"]
|
|
|
|
# ── Datetime helpers ──────────────────────────────────────────────────────────
|
|
def _parse_day(val) -> date:
|
|
if isinstance(val, date) and not isinstance(val, datetime):
|
|
return val
|
|
if isinstance(val, datetime):
|
|
return val.date()
|
|
s = str(val).strip().lower()
|
|
today = date.today()
|
|
if s in ("dnes", "today", ""):
|
|
return today
|
|
if s in ("zitra", "zítra", "tomorrow"):
|
|
return today + timedelta(days=1)
|
|
if s in ("vcera", "včera", "yesterday"):
|
|
return today - timedelta(days=1)
|
|
if s.startswith(("+", "-")) and s[1:].isdigit():
|
|
return today + timedelta(days=int(s))
|
|
return datetime.strptime(s, "%Y-%m-%d").date()
|
|
|
|
def _day_to_utc_range(d: date) -> tuple[str, str]:
|
|
start = datetime.combine(d, dtime.min).replace(tzinfo=PRAGUE_TZ)
|
|
end = datetime.combine(d, dtime.max.replace(microsecond=0)).replace(tzinfo=PRAGUE_TZ)
|
|
fmt = "%Y-%m-%dT%H:%M:%S.000Z"
|
|
return start.astimezone(tz.UTC).strftime(fmt), end.astimezone(tz.UTC).strftime(fmt)
|
|
|
|
def _to_utc_iso(dt: datetime) -> str:
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=PRAGUE_TZ)
|
|
return dt.astimezone(tz.UTC).strftime("%Y-%m-%dT%H:%M:%S.000Z")
|
|
|
|
def _resolve_calendar(cal: Optional[str]) -> str:
|
|
if cal is None:
|
|
return CALENDARS["vlado"]
|
|
if cal in CALENDARS:
|
|
return CALENDARS[cal]
|
|
if len(cal) == 36 and cal.count("-") == 4:
|
|
return cal
|
|
raise ValueError(f"Neznámý kalendář: {cal!r}. Použij: {list(CALENDARS)} nebo UUID.")
|
|
|
|
# ── MCP server ────────────────────────────────────────────────────────────────
|
|
mcp = FastMCP("medevio")
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# AGENDA
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
_AGENDA_QUERY = """
|
|
query Agenda_ListAll(
|
|
$calendarIds: [UUID!]!, $clinicSlug: String!,
|
|
$locale: Locale!, $since: DateTime!, $until: DateTime!
|
|
) {
|
|
reservations: listClinicReservations(
|
|
clinicSlug: $clinicSlug, calendarIds: $calendarIds,
|
|
since: $since, until: $until
|
|
) {
|
|
id start end note done color canceledAt calendarId
|
|
request {
|
|
id displayTitle(locale: $locale)
|
|
extendedPatient {
|
|
id name surname dob phone identificationNumber
|
|
insuranceCompanyObject { code shortName }
|
|
}
|
|
}
|
|
}
|
|
recurringReservations: listClinicRecurringReservations(
|
|
clinicSlug: $clinicSlug, calendarIds: $calendarIds,
|
|
since: $since, until: $until
|
|
) {
|
|
recurringReservation { id calendarId color note rrule { frequency interval dtstart tzid byweekday } }
|
|
instances { start end note color }
|
|
}
|
|
}"""
|
|
|
|
_CAL_NAME = {v: k for k, v in CALENDARS.items()}
|
|
|
|
def _parse_agenda_response(data: dict, cal_ids: list[str]) -> list[dict]:
|
|
parsed = []
|
|
|
|
for res in data.get("reservations") or []:
|
|
if res.get("canceledAt"):
|
|
continue
|
|
s = dtparser.isoparse(res["start"]).astimezone(PRAGUE_TZ)
|
|
e = dtparser.isoparse(res["end"]).astimezone(PRAGUE_TZ)
|
|
req = res.get("request") or {}
|
|
pat = req.get("extendedPatient") or {}
|
|
ins = pat.get("insuranceCompanyObject") or {}
|
|
parsed.append({
|
|
"typ": "pacient" if req.get("id") else "poznamka",
|
|
"kalendar": _CAL_NAME.get(res.get("calendarId"), res.get("calendarId")),
|
|
"start": s.strftime("%Y-%m-%d %H:%M"),
|
|
"end": e.strftime("%H:%M"),
|
|
"titul": req.get("displayTitle") or "",
|
|
"pacient": f"{pat.get('surname','')} {pat.get('name','')}".strip(),
|
|
"dob": pat.get("dob") or "",
|
|
"rc": pat.get("identificationNumber") or "",
|
|
"telefon": pat.get("phone") or "",
|
|
"pojistovna": ins.get("shortName") or "",
|
|
"poznamka": (res.get("note") or "").strip(),
|
|
"hotovo": bool(res.get("done")),
|
|
"reservation_id": res["id"],
|
|
"request_id": req.get("id") or "",
|
|
"patient_id": pat.get("id") or "",
|
|
"opakovana": False,
|
|
})
|
|
|
|
for rr in data.get("recurringReservations") or []:
|
|
rule = rr.get("recurringReservation") or {}
|
|
cal_id = rule.get("calendarId")
|
|
for inst in rr.get("instances") or []:
|
|
s = dtparser.isoparse(inst["start"]).astimezone(PRAGUE_TZ)
|
|
e = dtparser.isoparse(inst["end"]).astimezone(PRAGUE_TZ)
|
|
parsed.append({
|
|
"typ": "poznamka",
|
|
"kalendar": _CAL_NAME.get(cal_id, cal_id),
|
|
"start": s.strftime("%Y-%m-%d %H:%M"),
|
|
"end": e.strftime("%H:%M"),
|
|
"titul": "",
|
|
"pacient": "",
|
|
"dob": "", "rc": "", "telefon": "", "pojistovna": "",
|
|
"poznamka": (inst.get("note") or rule.get("note") or "").strip(),
|
|
"hotovo": False,
|
|
"reservation_id": rule.get("id"),
|
|
"request_id": "",
|
|
"patient_id": "",
|
|
"opakovana": True,
|
|
"rrule": rule.get("rrule"),
|
|
})
|
|
|
|
parsed.sort(key=lambda x: x["start"])
|
|
return parsed
|
|
|
|
|
|
@mcp.tool()
|
|
def get_agenda(
|
|
den: str = "dnes",
|
|
do: str = "",
|
|
kalendar: str = "",
|
|
) -> dict:
|
|
"""Vrátí agendu z Medevio kalendáře.
|
|
|
|
Args:
|
|
den: Datum začátku. Formáty: 'dnes', 'zitra', 'vcera', '+N', '-N', 'YYYY-MM-DD'.
|
|
Default 'dnes'.
|
|
do: Datum konce (volitelné). Stejné formáty jako 'den'. Prázdné = stejný den jako 'den'.
|
|
kalendar: 'vlado', 'manzelka', nebo UUID. Prázdné = oba kalendáře.
|
|
"""
|
|
try:
|
|
start_d = _parse_day(den)
|
|
end_d = _parse_day(do) if do.strip() else start_d
|
|
|
|
if kalendar.strip():
|
|
cal_ids = [_resolve_calendar(kalendar.strip())]
|
|
else:
|
|
cal_ids = list(CALENDARS.values())
|
|
|
|
fmt = "%Y-%m-%dT%H:%M:%S.000Z"
|
|
since = datetime.combine(start_d, dtime.min).replace(tzinfo=PRAGUE_TZ).astimezone(tz.UTC).strftime(fmt)
|
|
until = datetime.combine(end_d, dtime.max.replace(microsecond=0)).replace(tzinfo=PRAGUE_TZ).astimezone(tz.UTC).strftime(fmt)
|
|
|
|
data = _gql("Agenda_ListAll", _AGENDA_QUERY, {
|
|
"calendarIds": cal_ids,
|
|
"clinicSlug": CLINIC_SLUG,
|
|
"since": since,
|
|
"until": until,
|
|
"locale": "cs",
|
|
})
|
|
|
|
zaznamy = _parse_agenda_response(data, cal_ids)
|
|
return {
|
|
"od": start_d.isoformat(),
|
|
"do": end_d.isoformat(),
|
|
"pocet": len(zaznamy),
|
|
"zaznamy": zaznamy,
|
|
}
|
|
except Exception:
|
|
log(f"get_agenda chyba: {traceback.format_exc()}")
|
|
raise
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# ZÁPIS / MAZÁNÍ POZNÁMEK V KALENDÁŘI
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
@mcp.tool()
|
|
def zapis_poznamku(
|
|
poznamka: str,
|
|
den: str,
|
|
cas: str,
|
|
trvani_min: int = 5,
|
|
kalendar: str = "vlado",
|
|
color: str = "CHARCOAL",
|
|
) -> dict:
|
|
"""Vytvoří poznámku lékaře v Medevio kalendáři.
|
|
|
|
Args:
|
|
poznamka: Text poznámky.
|
|
den: Datum: 'dnes', 'zitra', 'YYYY-MM-DD', '+N' apod.
|
|
cas: Čas začátku ve formátu 'HH:MM' (lokální čas Praha).
|
|
trvani_min: Délka v minutách (default 5).
|
|
kalendar: 'vlado' (default), 'manzelka', nebo UUID.
|
|
color: Barva bloku (default 'CHARCOAL'). Další možnosti: RED, GREEN, BLUE, ORANGE, PURPLE aj.
|
|
"""
|
|
mutation = """
|
|
mutation CreateReservation_MakeReservationByDoctor(
|
|
$clinicSlug: String!, $color: ECRFIconColor, $note: String!, $timeSlotInput: TimeSlotInput!
|
|
) {
|
|
reservation: makeReservationByDoctor(
|
|
clinicSlug: $clinicSlug color: $color note: $note timeSlotInput: $timeSlotInput
|
|
) { id __typename }
|
|
}"""
|
|
|
|
try:
|
|
d = _parse_day(den)
|
|
t = datetime.strptime(cas.strip(), "%H:%M").time()
|
|
cal_id = _resolve_calendar(kalendar)
|
|
start_dt = datetime.combine(d, t).replace(tzinfo=PRAGUE_TZ)
|
|
end_dt = start_dt + timedelta(minutes=int(trvani_min))
|
|
|
|
data = _gql("CreateReservation_MakeReservationByDoctor", mutation, {
|
|
"clinicSlug": CLINIC_SLUG,
|
|
"color": color,
|
|
"note": poznamka,
|
|
"timeSlotInput": {
|
|
"calendarId": cal_id,
|
|
"start": _to_utc_iso(start_dt),
|
|
"end": _to_utc_iso(end_dt),
|
|
},
|
|
})
|
|
res_id = data["reservation"]["id"]
|
|
return {
|
|
"ok": True,
|
|
"reservation_id": res_id,
|
|
"den": d.isoformat(),
|
|
"cas": t.strftime("%H:%M"),
|
|
"trvani_min": trvani_min,
|
|
"poznamka": poznamka,
|
|
"kalendar": _CAL_NAME.get(cal_id, cal_id),
|
|
}
|
|
except Exception:
|
|
log(f"zapis_poznamku chyba: {traceback.format_exc()}")
|
|
raise
|
|
|
|
|
|
@mcp.tool()
|
|
def smaz_rezervaci(
|
|
reservation_id: str,
|
|
clinic_slug: str = CLINIC_SLUG,
|
|
) -> dict:
|
|
"""Zruší (cancel) jednorazovou rezervaci nebo poznámku lékaře.
|
|
|
|
Args:
|
|
reservation_id: UUID rezervace (z get_agenda → reservation_id).
|
|
clinic_slug: Slug kliniky (default mudr-buzalkova).
|
|
"""
|
|
mutation = """
|
|
mutation UpdateReservation_CancelReservationByDoctor(
|
|
$clinicSlug: String!, $reservationId: UUID!
|
|
) {
|
|
reservation: cancelReservationByDoctor(
|
|
clinicSlug: $clinicSlug, reservationId: $reservationId
|
|
) { id __typename }
|
|
}"""
|
|
|
|
try:
|
|
data = _gql("UpdateReservation_CancelReservationByDoctor", mutation, {
|
|
"clinicSlug": clinic_slug,
|
|
"reservationId": reservation_id,
|
|
})
|
|
return {"ok": True, "zruseno_id": data["reservation"]["id"]}
|
|
except Exception:
|
|
log(f"smaz_rezervaci chyba: {traceback.format_exc()}")
|
|
raise
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# POŽADAVKY
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
_POZADAVKY_QUERY = """
|
|
query ClinicRequestList2(
|
|
$clinicSlug: String!, $queueAssignment: QueueAssignmentFilter!,
|
|
$state: PatientRequestState, $pageInfo: PageInfo!, $locale: Locale!
|
|
) {
|
|
requestsResponse: listPatientRequestsForClinic2(
|
|
clinicSlug: $clinicSlug, queueAssignment: $queueAssignment,
|
|
state: $state, pageInfo: $pageInfo
|
|
) {
|
|
count
|
|
patientRequests {
|
|
id displayTitle(locale: $locale)
|
|
createdAt updatedAt doneAt removedAt
|
|
extendedPatient { id name surname identificationNumber phone }
|
|
lastMessage { createdAt }
|
|
}
|
|
}
|
|
}"""
|
|
|
|
@mcp.tool()
|
|
def get_pozadavky(
|
|
stav: str = "ACTIVE",
|
|
pocet: int = 50,
|
|
offset: int = 0,
|
|
) -> dict:
|
|
"""Vrátí seznam požadavků z Medevia.
|
|
|
|
Args:
|
|
stav: 'ACTIVE' (default) nebo 'DONE' (vyřízené).
|
|
pocet: Počet záznamů na stránku (default 50, max 100).
|
|
offset: Offset pro stránkování (default 0).
|
|
"""
|
|
try:
|
|
data = _gql("ClinicRequestList2", _POZADAVKY_QUERY, {
|
|
"clinicSlug": CLINIC_SLUG,
|
|
"queueAssignment": "ALL",
|
|
"state": stav.upper(),
|
|
"pageInfo": {"first": min(pocet, 100), "offset": offset},
|
|
"locale": "cs",
|
|
})
|
|
resp = data["requestsResponse"]
|
|
zaznamy = []
|
|
for r in resp.get("patientRequests") or []:
|
|
pat = r.get("extendedPatient") or {}
|
|
lm = r.get("lastMessage") or {}
|
|
zaznamy.append({
|
|
"id": r["id"],
|
|
"titul": r.get("displayTitle") or "",
|
|
"pacient": f"{pat.get('surname','')} {pat.get('name','')}".strip(),
|
|
"rc": pat.get("identificationNumber") or "",
|
|
"telefon": pat.get("phone") or "",
|
|
"patient_id": pat.get("id") or "",
|
|
"vytvoreno": r.get("createdAt") or "",
|
|
"aktualizovano": r.get("updatedAt") or "",
|
|
"dokonceno": r.get("doneAt") or "",
|
|
"posledni_zprava": lm.get("createdAt") or "",
|
|
})
|
|
return {
|
|
"stav": stav.upper(),
|
|
"celkem": resp.get("count", 0),
|
|
"offset": offset,
|
|
"pocet": len(zaznamy),
|
|
"pozadavky": zaznamy,
|
|
}
|
|
except Exception:
|
|
log(f"get_pozadavky chyba: {traceback.format_exc()}")
|
|
raise
|
|
|
|
|
|
_POZADAVEK_DETAIL_QUERY = """
|
|
query ClinicRequestDetail_GetPatientRequest2(
|
|
$clinicSlug: String!, $requestId: ID!, $isDoctor: Boolean!, $locale: Locale!
|
|
) {
|
|
request: getPatientRequest2(clinicSlug: $clinicSlug, requestId: $requestId) {
|
|
id doneAt doneBy { id name surname }
|
|
removedAt createdAt createdBy { id name surname }
|
|
displayTitle(locale: $locale) customTitle
|
|
clinicMedicalRecord clinicMedicalRecordVisibleToPatient
|
|
userNote evaluationResult
|
|
queue { id name }
|
|
substate
|
|
hasMobileApp
|
|
tags { id name }
|
|
extendedPatient {
|
|
id name surname dob identificationNumber phone email
|
|
insuranceCompanyObject { code name shortName }
|
|
note city street houseNumber
|
|
}
|
|
}
|
|
}"""
|
|
|
|
@mcp.tool()
|
|
def get_pozadavek(request_id: str) -> dict:
|
|
"""Vrátí detail jednoho požadavku.
|
|
|
|
Args:
|
|
request_id: UUID požadavku.
|
|
"""
|
|
try:
|
|
data = _gql("ClinicRequestDetail_GetPatientRequest2", _POZADAVEK_DETAIL_QUERY, {
|
|
"clinicSlug": CLINIC_SLUG,
|
|
"requestId": request_id,
|
|
"isDoctor": True,
|
|
"locale": "cs",
|
|
})
|
|
return data.get("request") or {}
|
|
except Exception:
|
|
log(f"get_pozadavek chyba: {traceback.format_exc()}")
|
|
raise
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# INTERNÍ KLINICKÉ POZNÁMKY K POŽADAVKU
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
_NOTES_GET_QUERY = """
|
|
query ClinicRequestNotes_Get($patientRequestId: String!) {
|
|
notes: getClinicPatientRequestNotes(requestId: $patientRequestId) {
|
|
id content createdAt updatedAt createdBy { id name surname }
|
|
}
|
|
}"""
|
|
|
|
_NOTE_UPDATE_MUTATION = """
|
|
mutation ClinicRequestNotes_Update($noteInput: UpdateClinicPatientRequestNoteInput!) {
|
|
updateClinicPatientRequestNote(noteInput: $noteInput) { id }
|
|
}"""
|
|
|
|
_NOTE_CREATE_MUTATION = """
|
|
mutation ClinicRequestNotes_Create($noteInput: CreateClinicPatientRequestNoteInput!) {
|
|
createClinicPatientRequestNote(noteInput: $noteInput) { id }
|
|
}"""
|
|
|
|
@mcp.tool()
|
|
def get_poznamky(request_id: str) -> dict:
|
|
"""Vrátí interní klinické poznámky k požadavku.
|
|
|
|
Args:
|
|
request_id: UUID požadavku.
|
|
"""
|
|
try:
|
|
data = _gql("ClinicRequestNotes_Get", _NOTES_GET_QUERY, {"patientRequestId": request_id})
|
|
return {"request_id": request_id, "poznamky": data.get("notes") or []}
|
|
except Exception:
|
|
log(f"get_poznamky chyba: {traceback.format_exc()}")
|
|
raise
|
|
|
|
|
|
@mcp.tool()
|
|
def uloz_poznamku(
|
|
request_id: str,
|
|
obsah: str,
|
|
prepend: bool = False,
|
|
) -> dict:
|
|
"""Vytvoří nebo aktualizuje interní klinickou poznámku k požadavku.
|
|
|
|
Pokud poznámka k požadavku neexistuje, vytvoří novou.
|
|
Pokud existuje, aktualizuje ji (buď nahradí celý obsah, nebo přidá text na začátek).
|
|
|
|
Args:
|
|
request_id: UUID požadavku.
|
|
obsah: Nový obsah poznámky (nebo text k přidání na začátek, pokud prepend=True).
|
|
prepend: Pokud True, přidá `obsah` na začátek existující poznámky (default False = přepíše).
|
|
"""
|
|
try:
|
|
notes_data = _gql("ClinicRequestNotes_Get", _NOTES_GET_QUERY, {"patientRequestId": request_id})
|
|
notes = notes_data.get("notes") or []
|
|
|
|
if notes:
|
|
note = notes[0]
|
|
note_id = note["id"]
|
|
new_content = (obsah + "\n" + (note["content"] or "")) if prepend else obsah
|
|
result = _gql("ClinicRequestNotes_Update", _NOTE_UPDATE_MUTATION, {
|
|
"noteInput": {"id": note_id, "content": new_content}
|
|
})
|
|
return {"ok": True, "akce": "update", "note_id": result["updateClinicPatientRequestNote"]["id"]}
|
|
else:
|
|
result = _gql("ClinicRequestNotes_Create", _NOTE_CREATE_MUTATION, {
|
|
"noteInput": {"requestId": request_id, "content": obsah}
|
|
})
|
|
return {"ok": True, "akce": "create", "note_id": result["createClinicPatientRequestNote"]["id"]}
|
|
except Exception:
|
|
log(f"uloz_poznamku chyba: {traceback.format_exc()}")
|
|
raise
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# PACIENTI
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
_SEARCH_QUERY = """
|
|
query Search($clinicSlug: String!, $locale: Locale!, $query: String!) {
|
|
results: search(clinicSlug: $clinicSlug, locale: $locale, query: $query) {
|
|
patients {
|
|
id name surname identificationNumber dob
|
|
insuranceCompanyObject { code shortName }
|
|
status isInClinic
|
|
}
|
|
}
|
|
}"""
|
|
|
|
@mcp.tool()
|
|
def hledej_pacienta(query: str) -> dict:
|
|
"""Vyhledá pacienta v Medeviu.
|
|
|
|
Args:
|
|
query: Jméno, příjmení, rodné číslo nebo jejich kombinace.
|
|
"""
|
|
try:
|
|
data = _gql("Search", _SEARCH_QUERY, {
|
|
"clinicSlug": CLINIC_SLUG,
|
|
"locale": "cs",
|
|
"query": query,
|
|
})
|
|
pacienti = data.get("results", {}).get("patients") or []
|
|
return {"query": query, "pocet": len(pacienti), "pacienti": pacienti}
|
|
except Exception:
|
|
log(f"hledej_pacienta chyba: {traceback.format_exc()}")
|
|
raise
|
|
|
|
|
|
_PATIENT_DETAIL_QUERY = """
|
|
query GetPatientDetail($clinicSlug: String!, $patientId: String!) {
|
|
patient: getPatientForClinic(clinicSlug: $clinicSlug, patientId: $patientId) {
|
|
id name surname identificationNumber sex dob
|
|
email phone status isInClinic hasMobileApp
|
|
note city street houseNumber createdAt
|
|
insuranceCompanyObject { code name shortName }
|
|
user { id email phone }
|
|
}
|
|
}"""
|
|
|
|
@mcp.tool()
|
|
def get_pacient(patient_id: str) -> dict:
|
|
"""Vrátí detail pacienta.
|
|
|
|
Args:
|
|
patient_id: UUID pacienta (z hledej_pacienta nebo get_agenda).
|
|
"""
|
|
try:
|
|
data = _gql("GetPatientDetail", _PATIENT_DETAIL_QUERY, {
|
|
"clinicSlug": CLINIC_SLUG,
|
|
"patientId": patient_id,
|
|
})
|
|
return data.get("patient") or {}
|
|
except Exception:
|
|
log(f"get_pacient chyba: {traceback.format_exc()}")
|
|
raise
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
if __name__ == "__main__":
|
|
log("MCP Medevio server spuštěn (FastMCP)")
|
|
mcp.run()
|