Construirea unui adaptor MCP cu Flask: declararea instrumentelor, rutare și depanare

Când conectezi un mediu de browser la un AI Agent, există un lucru inevitabil: instrumentele standard oferă operații primitive, dar acțiunile de business trebuie compuse de tine.

Să luăm un exemplu concret. Pentru a finaliza o sesiune de colectare, fluxul real este „deschide un mediu pentru o regiune, leagă ieșirea de rețea, fă o vizită de încălzire și confirmă că ieșirea funcționează”, iar abia apoi predai lucrul Agentului. Un MCP Server gata făcut oferă de regulă doar instrumente cu un singur pas: deschiderea mediului, navigare, clic și captură de ecran. Nimeni nu compune pentru tine secvența de mai sus. Agentul trebuie să o ordoneze de la zero de fiecare dată, ceea ce este lent, iar dacă un pas intermediar eșuează, tot el trebuie să decidă ce urmează.

Aici intervine stratul de adaptare: păstrezi la tine logica de compunere, validarea și starea, iar în exterior expui o singură acțiune de business.

Structura minimă funcțională

Un strat de adaptare funcțional are nevoie de doar trei părți: declararea instrumentelor, rutarea cererilor și execuția cu rezultat. Cu Flask, toate pot rula într-un singur proces.

Începe cu declararea instrumentelor; aceasta stabilește ce poate vedea Agentul.

# adapter/tools.py
TOOLS = [
    {
        "name": "prepare_environment",
        "description": "Pregătește un mediu disponibil pentru o regiune și returnează ID-ul mediului la final",
        "inputSchema": {
            "type": "object",
            "properties": {
                "region": {"type": "string", "description": "Regiunea de ieșire, de exemplu US-CA"},
                "purpose": {"type": "string", "description": "Etichetă de scop pentru reutilizare și statistici de cotă"},
                "timeout": {"type": "integer", "minimum": 10, "maximum": 120, "default": 60},
            },
            "required": ["region"],
            "additionalProperties": False,
        },
    },
    {
        "name": "open_page",
        "description": "Deschide o pagină în mediul specificat și așteaptă până devine interactivă",
        "inputSchema": {
            "type": "object",
            "properties": {
                "env_id": {"type": "string"},
                "url": {"type": "string"},
            },
            "required": ["env_id", "url"],
            "additionalProperties": False,
        },
    },
]

Routerul distribuie metodele JSON-RPC. Clientul cere mai întâi tools/list, apoi apelează tools/call după nume.

# adapter/app.py
from flask import Flask, request, jsonify
from adapter.tools import TOOLS
from adapter.runner import run_tool

app = Flask(__name__)

@app.post("/mcp")
def mcp():
    req = request.get_json(force=True)
    method, rid = req.get("method"), req.get("id")

    if method == "tools/list":
        return jsonify({"jsonrpc": "2.0", "id": rid, "result": {"tools": TOOLS}})

    if method == "tools/call":
        name = req["params"]["name"]
        args = req["params"].get("arguments", {})
        return jsonify({"jsonrpc": "2.0", "id": rid, "result": run_tool(name, args)})

    return jsonify({"jsonrpc": "2.0", "id": rid,
                    "error": {"code": -32601, "message": "method not found"}})

Stratul de execuție este singurul loc care atinge API-ul mediului și protocolul paginii. Aici se execută secvențele cu mai mulți pași și tot aici erorile sunt transformate într-un format uniform.

# adapter/runner.py
def run_tool(name, args):
    try:
        payload = HANDLERS[name](**args)
        return ok(payload)
    except ToolError as e:
        return fail(e.code, e.retryable, e.message)

Merită să stabilești devreme formatul de răspuns. Nu trimite Agentului excepțiile brute din nivelurile inferioare; va încerca să interpreteze acele șiruri și poate lua decizii ciudate. Definește o structură cu un cod de rezultat, de exemplu {"ok": false, "code": "env_unavailable", "retryable": true, "attempts": 3}. Astfel, Agentul trebuie să decidă doar două lucruri: se poate reîncerca și trebuie cazul predat unei persoane?

Cum proiectezi parametrii

Granularitatea instrumentelor trebuie stabilită după acțiunile de business, nu după API-uri. Dacă împachetezi fiecare endpoint de nivel jos ca instrument separat, practic nu ai făcut abstracție; Agentul trebuie în continuare să ordoneze singur pașii.

La parametri există o linie clară: ce trebuie să decidă modelul și ce trebuie să decidă stratul de adaptare. Regiunea, scopul și URL-ul țintă țin de primul grup și sunt completate de model. Portul de depanare, numele cozilor interne și alegerea pool-ului de medii țin de al doilea; nu le pune în schema, fiindcă mai devreme sau mai târziu modelul le va completa greșit.

Selectorii sunt o sursă frecventă de probleme. Când structura paginii se schimbă, apelurile cu selectori fixați în cod pot cădea în serie. E mai bine ca modelul să trimită ținte semantice, de exemplu un reper relativ stabil precum „butonul de autentificare”, iar maparea selectorilor să rămână în stratul de adaptare. Astfel, la o schimbare modifici un singur loc.

Parametrii numerici trebuie să aibă limite superioare. Dacă timeout-ul, numărul de reîncercări sau numărul de pagini nu au maximum în schema, modelul poate trimite o valoare foarte mare și poate transforma un apel într-o operație de peste zece minute. Orice instrument cu logică de buclă trebuie să aibă un punct final clar. Folosește un câmp precum max_pages pentru a limita lucrul, nu o regulă de tipul „continuă până când nu mai există date”.

Trebuie luată în calcul și idempotența. Fă apelantul să trimită un task_id; cererile repetate pot returna direct rezultatul anterior, evitând ca un retry al Agentului să creeze mediul de două ori.

Păstrează răspunsurile mici. Nu returna capturi de ecran în base64; returnează o referință la fișier. Pentru liste, include count și un indicator de trunchiere în loc să pui tot tabelul în context.

La depanare, verifică mai întâi aceste trei puncte

Cu stdio, JSON-RPC ocupă stdout, deci logurile nu trebuie scrise niciodată în stdout. Un singur print pus din reflex poate strica imediat parsarea protocolului și îți poate consuma mult timp la diagnosticare. Trimite logurile consecvent în stderr sau într-un fișier.

tools/list este primul punct de verificare. Dacă declarațiile nu au fost încărcate, niciun apel ulterior nu va avea loc. Verifică întâi numele instrumentelor, structura schema și dacă additionalProperties blochează vreun parametru.

Al doilea punct este un trace pas cu pas. Pentru fiecare apel, înregistrează task_id, un rezumat al parametrilor, durata și codul rezultatului. Când apare o problemă, poți vedea dacă s-a blocat la crearea mediului, la navigare sau la validare. Păstrează parametrii originali ai cazurilor eșuate pentru a le putea reda exact — un bug reproductibil se repară mult mai repede.

Al treilea punct este un set de fixtures fixe: o pagină de test cu conținut stabil și câțiva localizatori de elemente care nu se schimbă. Un smoke test după fiecare modificare a stratului de adaptare economisește mai mult timp decât orice verificare verbală.

Limitele permisiunilor

Stratul de adaptare este locul unde se concentrează cele mai multe permisiuni din întregul sistem: credențialele, mediile și operațiile pe pagină sunt toate în mâna lui. De aceea, limitele trebuie impuse aici.

Ține credențialele în configurația serverului, nu în parametrii instrumentelor și nici în contextul modelului. Separă instrumentele după nivelul de risc: cele doar pentru citire, cum sunt capturile și extragerea textului, sunt activate implicit; cele de scriere, precum clic, trimitere și ștergere, sunt dezactivate implicit și se activează temporar pentru o sarcină. Astfel, chiar dacă modelul ia o decizie greșită, impactul rămâne limitat.

Izolează mediile după scop. Conturi diferite și sarcini diferite trebuie să folosească medii diferite; nu amesteca utilizările într-un singur mediu. În scenarii cu mai multe conturi, crearea mediului, legarea ieșirii de rețea și menținerea stării pot fi delegate unui strat dedicat mediilor. Instrumente precum PurpleMark separă acest strat, iar adaptorul rămâne responsabil doar de compunerea logicii de business și validare.

Păstrează logurile de audit. Trebuie să poți exporta în funcție de timp ce sarcină a folosit ce mediu și ce instrumente de scriere au fost apelate. Când apare o problemă reală, această evidență este singura cale de a reconstrui procesul.

La final, limita fermă: stratul de adaptare nu trebuie să ofere nicio cale de ocolire a regulilor platformei. Acțiuni precum înregistrarea în masă, ocolirea verificărilor sau falsificarea identității nu trebuie împachetate ca instrumente și nici ascunse sub denumirea de instrumente interne. Odată ce un instrument este expus modelului, apelurile pot avea loc automat; nu există un punct sigur în care să le oprești înainte și nici nu poți anula efectele după.

Pentru versiunile protocolului și definițiile câmpurilor, consultă documentația oficială MCP.