notebookvb
This commit is contained in:
+835
@@ -0,0 +1,835 @@
|
||||
#!/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
|
||||
"""
|
||||
|
||||
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()
|
||||
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 neexistuje, vytvoří novou.
|
||||
Pokud existuje, aktualizuje ji (buď přepíše 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()
|
||||
Reference in New Issue
Block a user