"""MCP-server som exponerar RPA-verktyget för en AI-agent (Claude Code,
Claude Desktop, eller annan MCP-klient).
Kör:
python -m rpa.mcp_server # stdio-transport
Registrera i Claude Code:
claude mcp add rpa -- <sökväg-till-.venv-python> -m rpa.mcp_server
Registrera i Claude Desktop (claude_desktop_config.json):
"rpa": { "command": "<.venv-python>", "args": ["-m", "rpa.mcp_server"] }
OBS: MCP-servern och webui:t delar samma Chromium-profilmapp
(data/browser_profile/). Bara EN körning kan använda den åt gången - stäng ett
körande flöde i webui:t innan du kör ett via MCP, och tvärtom.
"""
from __future__ import annotations
import json
import logging
import sys
from mcp.server import MCPServer
from . import blocks, flows, runs, schema
from .flows import FlowNotFound, InvalidFlow, InvalidFlowName
from .migrate import flow_to_graph
logging.basicConfig(level=logging.INFO, stream=sys.stderr)
log = logging.getLogger("rpa.mcp")
mcp = MCPServer(
"rpa-remisser",
instructions=(
"Bygg och kör webbautomationsflöden. Anropa describe_steps() först för att "
"se alla nodtyper, fält och handtagsmodellen. Ett flöde är en graf: "
"nodes (platta noder) + edges (kanter). En 'call_flow'-nod kör ett annat "
"flöde eller en lokal subgraf (flödesnyckeln 'subflows') som ett steg. "
"Flöden sparas som JSON och körs i en synlig webbläsare mot en beständig "
"inloggningssession."
),
)
_MAX_LOG_LINES = 60
# --------------------------------------------------------------------------
# Schema
# --------------------------------------------------------------------------
@mcp.tool()
def describe_steps() -> dict:
"""Alla stegtyper, deras fält, villkorstyper och körningsinställningar.
Använd detta som referens innan du skapar eller ändrar ett flöde."""
return {
"step_types": schema.STEP_TYPES,
"condition_kinds": schema.CONDITION_KINDS,
"settings": schema.SETTINGS_SCHEMA,
"input_fields": schema.INPUT_FIELDS,
"close_modes": schema.CLOSE_MODES,
"reference_markdown": schema.step_reference_markdown(),
}
@mcp.resource("rpa://schema", mime_type="application/json")
def schema_resource() -> str:
return json.dumps(
{
"step_types": schema.STEP_TYPES,
"condition_kinds": schema.CONDITION_KINDS,
"settings": schema.SETTINGS_SCHEMA,
},
ensure_ascii=False,
indent=2,
)
# --------------------------------------------------------------------------
# Flöden (CRUD)
# --------------------------------------------------------------------------
@mcp.tool()
def list_flows() -> list[str]:
"""Namnen på alla sparade flöden."""
return flows.list_flows()
@mcp.tool()
def get_flow(name: str) -> dict:
"""Hämtar ett sparat flöde i grafform ({nodes, edges, start, settings})."""
try:
return flows.load_flow(name)
except FlowNotFound:
return {"error": f"Flödet '{name}' finns inte. Tillgängliga: {flows.list_flows()}"}
except InvalidFlowName as exc:
return {"error": str(exc)}
def _as_graph(nodes, edges, start, settings, steps, subflows=None) -> dict:
"""Bygg ett flödesdict från antingen nodes/edges eller det gamla steps-formatet."""
if nodes is not None:
g = {"format": "graph", "nodes": nodes, "edges": edges or [],
"start": start, "settings": settings or {}}
if subflows:
g["subflows"] = subflows
return g
return flow_to_graph({"steps": steps or [], "settings": settings or {}})
@mcp.tool()
def save_flow(
name: str,
nodes: list | None = None,
edges: list | None = None,
start: str | None = None,
settings: dict | None = None,
steps: list | None = None,
subflows: list | None = None,
) -> dict:
"""Skapar eller skriver över ett flöde. Ange grafen som 'nodes' + 'edges'
(se describe_steps för nodtyper och handtag). 'subflows' = lokala namngivna
subgrafer som call_flow-noder kan anropa (target '#<id>'). Det gamla
'steps'-formatet accepteras också och migreras. Returnerar valideringsfel
utan att spara om flödet är ogiltigt."""
g = _as_graph(nodes, edges, start, settings, steps, subflows)
errors = schema.validate_graph(g, strict=True)
if errors:
return {"ok": False, "errors": errors}
try:
flows.save_flow(name, g, strict=True)
except InvalidFlowName as exc:
return {"ok": False, "errors": [str(exc)]}
except InvalidFlow as exc:
return {"ok": False, "errors": exc.errors}
return {"ok": True, "name": name, "node_count": len(g["nodes"])}
@mcp.tool()
def delete_flow(name: str) -> dict:
"""Tar bort ett sparat flöde."""
try:
existed = flows.delete_flow(name)
except InvalidFlowName as exc:
return {"ok": False, "error": str(exc)}
return {"ok": True, "existed": existed}
@mcp.tool()
def validate_flow(
nodes: list | None = None,
edges: list | None = None,
start: str | None = None,
settings: dict | None = None,
steps: list | None = None,
subflows: list | None = None,
) -> dict:
"""Validerar ett flöde (grafform) utan att spara det."""
g = _as_graph(nodes, edges, start, settings, steps, subflows)
errors = schema.validate_graph(g, strict=True)
return {"valid": not errors, "errors": errors}
@mcp.tool()
def list_blocks() -> list[dict]:
"""Återanvändbara flödesdelar (namn, beskrivning, antal steg)."""
return blocks.list_blocks()
@mcp.tool()
def get_block(name: str) -> dict:
"""Hämtar ett blocks steg - klistra in dem i ett flöde med save_flow."""
try:
return blocks.load_block(name)
except blocks.BlockNotFound:
return {"error": f"Blocket '{name}' finns inte. Tillgängliga: {[b['name'] for b in blocks.list_blocks()]}"}
except blocks.InvalidBlockName as exc:
return {"error": str(exc)}
# --------------------------------------------------------------------------
# Körning
# --------------------------------------------------------------------------
@mcp.tool()
def run_flow(
name: str,
close_mode: str = "always",
initial_vars: dict | None = None,
input_set: str | None = None,
) -> dict:
"""Startar en körning av ett sparat flöde i bakgrunden. close_mode:
always | on_success | never. Returnerar ett run_id - poll:a get_run(run_id)
för status och logg.
initial_vars: seedar variabler (t.ex. för flödets deklarerade 'inputs').
input_set: namnet på ett sparat värde-set i flödet - dess värden används
som grund, initial_vars läggs ovanpå."""
if close_mode not in schema.CLOSE_MODES:
close_mode = "always"
try:
flow = flows.load_flow(name)
except FlowNotFound:
return {"error": f"Flödet '{name}' finns inte."}
except InvalidFlowName as exc:
return {"error": str(exc)}
resolved: dict = {}
for inp in flow.get("inputs", []):
if isinstance(inp, dict) and inp.get("name") and inp.get("default") not in (None, ""):
resolved[str(inp["name"])] = str(inp["default"])
if input_set:
sets = flow.get("input_sets", {})
if input_set not in sets:
return {"error": f"Värde-set '{input_set}' finns inte. Tillgängliga: {list(sets)}"}
resolved.update({str(k): str(v) for k, v in sets[input_set].items()})
if initial_vars:
resolved.update({str(k): str(v) for k, v in initial_vars.items()})
run_id = runs.start_run(
flow,
close_mode=close_mode,
settings=flow.get("settings", {}),
initial_vars=resolved,
)
return {"run_id": run_id, "status": "running", "initial_vars": resolved}
@mcp.tool()
def get_run(run_id: str) -> dict:
"""Status, logg-svans och variabler för en körning."""
run = runs.get_run(run_id)
if run is None:
return {"error": f"Okänt run_id: {run_id}"}
log_lines = run["logs"]
return {
"status": run["status"],
"browser_open": run["browser_open"],
"trace_ready": run["trace_ready"],
"log_tail": log_lines[-_MAX_LOG_LINES:],
"log_truncated": len(log_lines) > _MAX_LOG_LINES,
"vars": run["vars"],
}
@mcp.tool()
def close_run_browser(run_id: str) -> dict:
"""Stänger webbläsaren för en körning som hålls öppen (close_mode never/on_success)."""
return {"ok": runs.close_browser(run_id)}
def main() -> None:
log.info("Startar RPA MCP-server (stdio)")
mcp.run(transport="stdio")
if __name__ == "__main__":
main()