> **ELAVULT (deprecated)** — a Linear-integrációról szóló szakasz elavult (Plane-re váltás). # MCP Integrációs Minták Fejlesztői Környezetekhez A Model Context Protocol (MCP) a mesterséges intelligencia asszisztensek és fejlesztői eszközök összekapcsolásának alapvető szabványává vált, amely kifinomult munkafolyamatokat tesz lehetővé a beszélgetési AI és a strukturált vállalati rendszerek között. **A production telepítések 20,5%-kal gyorsabb feladatvégzést és 100%-os sikerességi arányt jelentenek** a nem MCP implementációkhoz képest, miközben 19,2%-kal csökkentik az API hívások számát az intelligens kontextuskezelés révén. ## Alapvető kommunikációs architektúra Az MCP háromszintű architektúrán működik, amely a **JSON-RPC 2.0**-t használja kommunikációs alapként. A protokoll szabványosított interakciókat tesz lehetővé az AI gazdák (Claude Code, Cursor IDE), az MCP kliensek protokoll közvetítőkként való működése, és az MCP szerverek külső eszköz képességeinek közzététele között. Ez az architektúra N×M problémáról N+M problémára csökkenti az integráció összetettségét, az MCP-t az "USB-C az AI alkalmazásokhoz" pozícióba helyezve. ### Protokoll inicializálás és képesség egyeztetés Minden MCP kapcsolat egy strukturált kézfogással kezdődik, amely protokoll verzió kompatibilitást és képesség cserét hoz létre. A kliens verzió egyeztetéssel kezd, miközben deklarálja a képességeit: ```json { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2025-06-18", "capabilities": { "elicitation": {}, "sampling": {} }, "clientInfo": { "name": "ExampleHostApp", "version": "1.2.0" } } } ``` A szerver a támogatott képességeivel és eszköz ajánlataival válaszol, létrehozva a további interakciók szerződését: ```json { "jsonrpc": "2.0", "id": 1, "result": { "protocolVersion": "2025-06-18", "capabilities": { "tools": {"listChanged": true}, "resources": {"supported": true, "subscribe": false}, "prompts": {"supported": false} }, "serverInfo": { "name": "SampleMCPServer", "version": "0.1.0" } } } ``` Ez a képesség egyeztetés dinamikus eszköz felfedezést tesz lehetővé és biztosítja a kompatibilitást a különböző MCP implementációk között, a `listChanged` képesség lehetővé teszi a szervereknek új eszközök dinamikus regisztrálását futás közben. ## Fejlesztői munkafolyamat integrációs példák ### GitHub munkafolyamat automatizálás MCP-n keresztül A Claude Code a GitHubbal specializált MCP szervereken keresztül integrálódik, amelyek a repository műveleteket szabványosított eszközökként teszik elérhetővé. Egy teljes issue-tól-pull-request munkafolyamat bemutatja a protokoll gyakorlati erejét: **Felhasználói kérés**: "Implementálj JWT autentikációt a user service-hez a nyitott fejlesztési issue-k alapján" **LLM feldolgozási fázis**: 1. **Kontextus elemzés**: Claude elemzi a kérést, azonosítja a kulcs komponenseket (JWT, autentikáció, user service) 2. **Eszköz tervezés**: Meghatározza a szükséges GitHub műveletek sorrendjét (issue keresés → kódbázis elemzés → implementáció → PR létrehozás) 3. **Paraméter generálás**: A természetes nyelvet strukturált eszköz argumentumokká konvertálja **Issue elemzés és kód implementáció**: ```json { "jsonrpc": "2.0", "id": 7, "method": "tools/call", "params": { "name": "fetch_github_issues", "arguments": { "repo": "company/authentication-service", "state": "open", "labels": ["enhancement"] } } } ``` **LLM utó-feldolgozás**: - **Követelmények szintézise**: Claude feldolgozza a visszakapott issue adatokat, kinyeri a közös mintákat és követelményeket - **Architektúrális döntések**: A kódbázis elemzés alapján dönt a JWT implementációs megközelítésről (middleware, service réteg, stb.) - **Kód generálás**: Implementációs fájlokat hoz létre megfelelő hibakezeléssel, teszteléssel és dokumentációval - **Integrációs tervezés**: Meghatározza, hogyan integrálódik az új auth rendszer a meglévő szolgáltatásokkal A GitHub MCP szerver feldolgozza ezt a kérést és strukturált issue adatokat ad vissza, amelyeket a Claude Code aztán a követelmények megértésére és megoldások implementálására használ. **A Microsoft Azure DevOps csapata jelentése szerint több száz manuális teszt esetet konvertáltak automatizált szkriptekké ezzel a mintával**, a fejlesztők megjegyezték: "Szinte varázslatnak tűnt – leírni egy tesztet egyszerű angol nyelven és kapni egy futtatható automatizált szkriptet cserébe!" ### Linear projekt menedzsment integráció A Linear integráció MCP-n keresztül intelligens feladat vezénylést tesz lehetővé automatizált prioritás hozzárendeléssel és csapat koordinációval: **Felhasználói kérés**: "Hozz létre egy magas prioritású feladatot a JWT autentikáció implementálásához és rendeld hozzá a legjobb fejlesztőhöz az auth munkához" **LLM érvelési folyamat**: 1. **Feladat dekompozíció**: Claude a JWT implementációt részfeladatokra bontja (tervezés, kódolás, tesztelés, telepítés) 2. **Prioritás értékelés**: Elemzi a projekt kontextust, hogy megállapítsa, megfelelő-e a "magas" prioritás a jelenlegi sprint terhelés alapján 3. **Csapat elemzés**: Feldolgozza a csapattagok adatait az autentikációs szakértelemmel rendelkező fejlesztők azonosítására, figyelembe véve a jelenlegi terhelést és múltbeli teljesítményt 4. **Kontextuális kapcsolás**: Összekapcsolja a Linear issue-t a kapcsolódó GitHub repository-kkal, dokumentációval és monitoring rendszerekkel ```json { "jsonrpc": "2.0", "id": 8, "method": "tools/call", "params": { "name": "linear_create_issue", "arguments": { "title": "JWT autentikációs rendszer implementálása", "description": "Biztonságos autentikáció hozzáadása teljesítmény elemzés alapján", "priority": "high", "assignee": "PROMPT_ARGUMENT: legjobb fejlesztő auth munkához csapat szakértelem alapján" } } } ``` **LLM létrehozás utáni elemzés**: - **Függőségi térképezés**: Claude automatikusan azonosítja a kapcsolódó feladatokat és javasol kapcsolási vagy blokkoló kapcsolatokat - **Becslési intelligencia**: Story point becsléseket ad hasonló befejezett feladatok alapján - **Kockázat értékelés**: Megjelöli a potenciális blokkolókat vagy függőségeket, amelyek befolyásolhatják az időkeretet Ez a minta **AI-vezérelt projekt koordinációt** tesz lehetővé, ahol Claude elemzi a projekt kontextust, csapat képességeket és jelenlegi munkaterhelést intelligens hozzárendelési döntések meghozatalához. A rendszer automatikusan kapcsolja az issue-kat a releváns GitHub repository-khoz és monitoring rendszerekhez kereszt-MCP szerver kommunikáción keresztül. ### Playwright teszt munkafolyamat automatizálás A Playwright MCP integráció a teszt munkafolyamatokat manuális szkript készítésről beszélgetési teszt generálásra alakítja: **Felhasználói kérés**: "Hozz létre átfogó teszteket az autentikációs flow-hoz, beleértve a szélsőséges eseteket" **LLM teszt stratégia fejlesztés**: 1. **Flow elemzés**: Claude feltérképezi a teljes autentikációs utazást (bejelentkezés, token validáció, session kezelés, kijelentkezés) 2. **Szélsőséges esetek azonosítása**: Automatikusan azonosítja a potenciális hiba szcenáriókat (érvénytelen hitelesítő adatok, lejárt tokenek, hálózati hibák, egyidejű sessionök) 3. **Teszt adat stratégia**: Meghatározza, milyen teszt felhasználók és szcenáriók szükségesek 4. **Állítás tervezés**: Dönt a megfelelő validációs pontokról és elvárt viselkedésekről **Generált implementáció**: ```javascript // Generált Playwright teszt MCP-n keresztül test('Autentikációs flow validáció', async ({ page }) => { const testUser = await mcp.getTestUser(browserName); await page.goto('/login'); await page.fill('[data-testid="username"]', testUser.email); await page.fill('[data-testid="password"]', testUser.password); await expect(page).toHaveURL('/dashboard'); }); ``` **LLM minőségbiztosítási feldolgozás**: - **Lefedettség elemzés**: Claude biztosítja, hogy a tesztek lefedik a boldog utat, hiba szcenáriókat és határ feltételeket - **Teljesítmény megfontolások**: Megfelelő várakozásokat és timeout kezelést ad hozzá - **Akadálymentesség validáció**: Beépíti az ARIA címke tesztelést és billentyűzetes navigációt - **Böngészők közötti kompatibilitás**: Generál böngésző-specifikus teszt variációkat, amikor szükséges - **Karbantartási intelligencia**: Olyan teszteket hoz létre, amelyek ellenállnak a kisebb UI változásoknak **Valós világbeli hatás**: Az Azure DevOps csapatok óráról percekre csökkentették a teszt készítési időt, a generált tesztek olyan szélsőséges eseteket fedtek le, amelyeket a manuális tesztelés gyakran kihagyott. Egy csapat kritikus hibát fedezett fel a keresési funkcióban ("Star Wars" keresése elrontotta a UI-t), amely hónapokig rejtve maradt a manuális tesztelésben. **LLM folyamatos tanulás**: Claude tanul a teszt végrehajtási eredményekből, automatikusan javasol javításokat a teszt stratégiákhoz és azonosítja a mintákat a hibákban különböző alkalmazások között. ## Autentikáció és session kezelési minták ### OAuth 2.1 kötelező implementáció Az **MCP 2025 specifikáció kötelezővé teszi az OAuth 2.1-et** minden production szerver számára, jelentős érettséget képviselve a vállalati szintű biztonság felé. Kulcs követelmények: - **PKCE (Proof Key for Code Exchange) kötelező** minden klienshez S256 kód kihívás módszerrel - **Dynamic Client Registration (DCR)** támogatás RFC7591 szerint a zökkenőmentes kliens bevezetéshez - **Authorization Server Metadata** megfelelőség RFC8414 szerint az endpoint felfedezéshez ```http # Authorizációs kérés PKCE-vel GET /oauth/authorize? response_type=code& client_id=mcp-client& code_challenge=E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM& code_challenge_method=S256& resource=https://mcp.example.com # Token csere POST /oauth/token Content-Type: application/x-www-form-urlencoded grant_type=authorization_code& code=SplxlOBeZQQYbYS6WxSbIA& code_verifier=dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk& resource=https://mcp.example.com ``` ### Session életciklus és elosztott kihívások Az MCP session kezelés egyedi kihívásokkal néz szembe az elosztott környezetekben. A sessionök az inicializációs fázisban jönnek létre és több eszköz meghívás során fenn kell tartani őket: ```http POST /mcp HTTP/1.1 Content-Type: application/json Mcp-Session-Id: cb7ed84c-8f2f-4109-b571-2f9fb025a5c2 { "jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": { "name": "database_query", "arguments": {"query": "SELECT * FROM users WHERE active=true"} } } ``` **Production telepítési kihívás**: Multi-worker környezetek (Kubernetes, Docker Swarm) session affinitási problémákba ütköznek, ahol az egyik workerben létrehozott sessionök elérhetetlenné válnak a másikból. Megoldások: - **Ragadós sessionök** NGINX Plus dinamikus térképezéssel - **Megosztott állapot tárolás** Redis klaszterekkel bérlő izolációval - **Állapot nélküli tervezés** JWT-alapú session tokenek használatával NGINX Plus konfiguráció session ragadáshoz: ```nginx upstream mcp_servers { sticky learn create=$upstream_cookie_jsessionid lookup=$cookie_jsessionid zone=client_sessions:1m; server backend1:8080; server backend2:8080; } ``` ## Hibakezelés és ellenállóság mechanizmusok ### Átfogó hiba osztályozási keretrendszer Az MCP háromszintű hiba modellt implementál, amely strukturált hiba válaszokat biztosít, lehetővé téve az intelligens kliens viselkedést: **Szállítási szintű hibák**: Hálózati timeout-ok, autentikációs hibák, kapcsolat létrehozási problémák **Protokoll szintű hibák**: Szabványosított JSON-RPC hibakódok specifikus kliens akció útmutatással **Alkalmazás szintű hibák**: Eszköz-specifikus hibák kontextuális hiba információval ```json { "jsonrpc": "2.0", "id": "request-123", "error": { "code": -32802, "message": "Erőforrás átmenetileg nem elérhető", "data": { "retry_after": 30, "resource_id": "database-primary", "suggested_action": "retry_with_backoff" } } } ``` ### Circuit breaker és újrapróbálkozási minták A production MCP implementációk kifinomult ellenállóság mintákat igényelnek a részleges hibák és szolgáltatás degradációs kezeléshez: ```python class CircuitBreaker: def __init__(self, failure_threshold=3, recovery_timeout=30): self.failure_count = 0 self.failure_threshold = failure_threshold self.is_open = False self.last_failure_time = None async def call(self, func): if self.is_open: if time.time() - self.last_failure_time > self.recovery_timeout: self.is_open = False # Fél-nyitott állapot self.failure_count = 0 else: raise Exception("Circuit breaker nyitva van") try: result = await func() self.failure_count = 0 # Reset sikernél return result except Exception as e: self.failure_count += 1 self.last_failure_time = time.time() if self.failure_count >= self.failure_threshold: self.is_open = True raise ``` **Exponenciális visszalépés zajjal** megelőzi a thundering herd problémákat: ```python async def retry_with_backoff(operation, max_attempts=3): for attempt in range(max_attempts): try: return await operation() except Exception as e: if attempt == max_attempts - 1: raise delay = min(2 ** attempt + random.uniform(0, 1), 60) await asyncio.sleep(delay) ``` ## Teljesítmény optimalizálás és benchmarkolás ### Valós világbeli teljesítmény adatok **Twilio MCP Server benchmark eredmények** konkrét teljesítmény metrikákat biztosítanak production telepítésekhez: - **Sebesség javulás**: 20,5%-kal gyorsabb feladat befejezés a nem MCP implementációkhoz képest - **API hatékonyság**: 19,2%-os csökkentés a szükséges API hívásokban intelligens kontextus cache-elés révén - **Sikerességi arány**: 100%-os feladat sikerességi arány vs 92,3% hagyományos integrációkkal - **Token költség hatás**: 27,5%-os növekedés token használatban a javított kontextus miatt, de **28,5%-kal több cache olvasás** és **53,7%-kal több cache írás** hatékony optimalizálási stratégiákat jelez ### Kontextus optimalizálási stratégiák Az MCP szerverek kifinomult cache-elést implementálnak a token fogyasztás minimalizálásához a kontextus gazdagság fenntartása mellett: ```python @lru_cache(maxsize=1000) async def database_tool(query_hash: str, query: str): cached_result = await redis_client.get(f"db:query:{query_hash}") if cached_result: return json.loads(cached_result) result = await execute_database_query(query) await redis_client.setex( f"db:query:{query_hash}", 300, # 5 perces TTL json.dumps(result) ) return result ``` ### Szállítási réteg teljesítmény megfontolások A különböző MCP szállítási mechanizmusok eltérő teljesítmény karakterisztikákat mutatnak: | Szállítás | Késleltetés (p99) | Áteresztőképesség | Használati eset | |-----------|------------------|-------------------|-----------------| | STDIO (helyi) | 1ms | 200K ops/s | Fejlesztés, helyi eszközök | | HTTP + SSE | 15ms | 50K ops/s | Távoli szerverek, web integráció | | WebSocket | 5ms | 100K ops/s | Valós idejű, kétirányú kommunikáció | **HTTP kapcsolat pooling** jelentősen javítja a teljesítményt távoli MCP szerverekhez: ```python async with httpx.AsyncClient( timeout=httpx.Timeout(30.0, connect=5.0), limits=httpx.Limits(max_connections=10, max_keepalive_connections=5) ) as client: response = await client.post('/mcp', json=request_payload) ``` ## Haladó eszköz láncolás és vezénylés ### Szekvenciális eszköz láncolás eredmény átadással A modern MCP implementációk kifinomult eszköz láncolást támogatnak, ahol az egyik eszköz kimenetei automatikusan a következő eszközök bemenetei lesznek: ```json { "tools": [ { "name": "analyze_codebase", "arguments": {"repository": "company/api-service"} }, { "name": "generate_tests", "arguments": { "code_analysis": "PROMPT_ARGUMENT: használd a komplexitási metrikákat az előző elemzésből" } }, { "name": "create_pull_request", "arguments": { "tests": "PROMPT_ARGUMENT: a teszt fájlok a 2. lépésből", "title": "PROMPT_ARGUMENT: leíró cím az elemzés alapján" } } ] } ``` Ez a minta **60-80%-kal csökkenti a token használatot** a szekvenciális egyedi eszköz hívásokhoz képest, miközben atomi multi-eszköz műveleteket tesz lehetővé teljes visszagörgetési képességekkel. ### Dinamikus képesség kompozíció A vállalati MCP telepítések növekvően használnak dinamikus eszköz regisztrációt a változó adatforrások és üzleti követelményekhez való alkalmazkodáshoz: ```python class DynamicToolRegistry: async def connect_to_data_source(self, data_source: DataSourceConfig): schema = await self.analyze_data_schema(data_source) generated_tools = [] for entity in schema.entities: crud_tools = self.generate_crud_tools(entity, data_source) for tool in crud_tools: await self.register_tool(tool, data_source.security_context) generated_tools.append(tool.name) await self.notify_list_changed() return generated_tools ``` Ez a megközelítés **kontextus-tudatos eszköz generálást** tesz lehetővé, ahol az MCP szerverek automatikusan létrehozzák a megfelelő eszközöket a kapcsolódó rendszerek alapján, csökkentve a manuális eszköz konfigurációs igényt és javítva a rendszer alkalmazkodóképességét. ### Multi-agent munkafolyamat vezénylés A haladó MCP implementációk több specializált agenst koordinálnak kifinomult vezénylési mintákon keresztül: - **Agent átadások** teljes kontextus megőrzéssel az átmenetek során - **Elosztott döntéshozatal** ahol több agent hozzájárul a komplex problémamegoldáshoz - **Hibatűrő koordináció** automatikus failover és helyreállítási mechanizmusokkal A **mcp-agent keretrendszer** ezt a mintát mutatja be a gyakorlatban: ```python from mcp_agent.workflows.orchestrator import Orchestrator orchestrator = Orchestrator( worker_agents=[data_analyst, report_generator, reviewer_agent], plan_type="iterative", error_handling="compensating_transactions" ) result = await orchestrator.generate_str( "Elemezd a Q4 eladási adatokat, generálj vezetői összefoglalót és küldd jóváhagyásra" ) ``` ## Production telepítés és vállalati minták ### Biztonság és megfelelőség architektúra A production MCP telepítések átfogó biztonsági intézkedéseket igényelnek az autentikáció, authorizáció és adatvédelem kezeléséhez: **Multi-réteg autentikáció**: - OAuth 2.1 PKCE-vel kliens autentikációhoz - mTLS szolgáltatások közötti kommunikációhoz - SPIFFE/SPIRE munkaterhelés identitás kezeléshez - Képesség-alapú hozzáférés kontroll finom szemcsés engedélyekhez **GDPR/CCPA megfelelőség** architektúrális minta szerint változik: | Minta | Adat törlés | Helyi vezérlés | Audit nyomvonal | |-------|-------------|----------------|-----------------| | Szerver oldali cache | Közvetlen kulcs törlés | Redis klaszter elhelyezés | Teljes naplózás | | Kliens pointerek | Tárolás törlés szükséges | CDN geo-korlátozások | Hozzáférés naplózás | | Állapottal rendelkező payload-ok | Komplex kliens oldali adat | Kliens tárolás | Korlátozott láthatóság | ### Konténer vezénylés és skálázás A Kubernetes telepítések horizontális skálázást és hibatűrést tesznek lehetővé: ```yaml apiVersion: apps/v1 kind: StatefulSet metadata: name: mcp-server-cluster spec: serviceName: mcp-server replicas: 3 template: spec: containers: - name: mcp-server image: mcp-server:latest env: - name: STATE_STORE value: "redis://redis-cluster:6379" - name: OAUTH_ENDPOINT value: "https://auth.company.com" resources: requests: memory: "512Mi" cpu: "500m" limits: memory: "1Gi" cpu: "1000m" ``` **Teljesítmény monitoring** kulcs metrikákra fókuszál: - Kérés késleltetés percentilisei (p50, p95, p99) - Eszköz végrehajtási sikerességi arányok - Token fogyasztási minták - Cache találat/miss arányok - Aktív session számok ## Gyakori hiba szcenáriók és megoldások ### Autentikáció és session hibák **Session affinitási problémák** terheléselosztott környezetekben: ``` Hiba: "Nem található session ID-hoz: cb7ed84c-8f2f-4109-b571-2f9fb025a5c2" Alapok: Session létrehozva A workerben, kérés B workerhez irányítva ``` **Megoldási minták**: - Redis-alapú megosztott session tárolás implementálása - JWT tokenek használata állapot nélküli session kezeléshez - Terheléselosztó ragadós sessionök konfigurálása ### Eszköz végrehajtási timeout-ok és erőforrás korlátok **Erőforrás kimerülés kezelés**: ```json { "jsonrpc": "2.0", "id": "req-456", "error": { "code": -32801, "message": "Tartalom túl nagy", "data": { "max_size": 1048576, "actual_size": 2097152, "suggested_action": "chunk_request" } } } ``` **Rate limiting válaszok** irányítják a kliens viselkedést: ```http HTTP/1.1 429 Too Many Requests Retry-After: 30 X-RateLimit-Limit: 100 X-RateLimit-Remaining: 0 X-RateLimit-Reset: 1640995260 { "error": { "code": 429, "message": "Rate limit túllépve", "retry_after": 30 } } ``` ## Jövőbeli megfontolások és feltörekvő minták Az **MCP ökoszisztéma folytatja a gyors evolúciót** vállalati igényeket kielégítő feltörekvő mintákkal: - **Kvantum számítástechnika integráció** ultra-gyors kontextus feldolgozáshoz specializált alkalmazásokban - **AI-vezérelt optimalizálás** önmaga tanuló késleltetés csökkentést és erőforrás allokációt lehetővé téve - **Neuromorphic számítástechnikai architektúrák** hatékony AI-eszköz interakciós mintákhoz tervezve A **token költség optimalizálás** kulcs fókusz marad, az implementációk mutatják az utat a csökkentett overhead felé a kifinomult kontextus kezelés fenntartása mellett. A jelenlegi telepítések demonstrálják, hogy bár az MCP 25-30%-os token overhead-et bevezet, az operációs előnyök (gyorsabb befejezés, magasabb sikerességi arányok, csökkentett fejlesztési idő) erős ROI indoklást biztosítanak. ## Következtetés Az MCP kísérleti protokollból production-kész szabvánnyá érett, amely kifinomult AI-eszköz integrációkat tesz lehetővé. **A valós világbeli telepítések mérhető javulásokat mutatnak**: 20%+ sebesség nyereségek, 100%-os feladat sikerességi arányok és jelentős csökkentés a fejlesztési időben. A protokoll evolúciója a kötelező OAuth 2.1 autentikáció és kifinomult vezénylési minták felé a vállalati AI munkafolyamatok alapvető szabványaként pozicionálja. Az MCP-vel való siker megköveteli a **háromszintű architektúra** megértését, a **megfelelő autentikáció és session kezelés** implementálását, az **ellenálló hibakezelés** tervezését és a **nagy léptékű teljesítmény** optimalizálását. Az itt dokumentált minták technikai alapot biztosítanak production-kész MCP integrációk építéséhez, amelyek áthidalják a beszélgetési AI-t és a strukturált vállalati rendszereket, lehetővé téve az AI-vezérelt fejlesztői környezetek következő generációját. ## Gyakorlati implementációs útmutató ### MCP szerver felállítás lépésről lépésre **1. Alapvető MCP szerver inicializálás** ```python from mcp.server import Server from mcp.types import Tool, TextContent import asyncio # MCP szerver példány létrehozása app = Server("Hungarian-Dev-Tools") @app.list_tools() async def list_tools(): return [ Tool( name="create_hungarian_docs", description="Magyar nyelvű dokumentáció generálása", inputSchema={ "type": "object", "properties": { "project_name": {"type": "string"}, "language": {"type": "string", "default": "hu"} } } ) ] @app.call_tool() async def call_tool(name: str, arguments: dict): if name == "create_hungarian_docs": return TextContent( type="text", text=f"Magyar dokumentáció generálva: {arguments['project_name']}" ) # Szerver indítása if __name__ == "__main__": asyncio.run(app.run()) ``` **2. Claude Code konfiguráció** ```json { "mcpServers": { "hungarian-dev-tools": { "command": "python", "args": ["hungarian_mcp_server.py"], "env": { "PYTHONPATH": "/path/to/mcp/server" } } } } ``` ### Hibakeresési és monitoring technikák **MCP kommunikáció naplózása** ```python import logging from mcp.server.stdio import stdio_server # Részletes logging beállítása logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('mcp_server.log'), logging.StreamHandler() ] ) logger = logging.getLogger("hungarian-mcp") @app.call_tool() async def call_tool(name: str, arguments: dict): logger.info(f"Eszköz hívás: {name} argumentumokkal: {arguments}") try: result = await process_tool_call(name, arguments) logger.info(f"Sikeres válasz: {len(str(result))} karakter") return result except Exception as e: logger.error(f"Hiba történt: {e}") raise ``` **Teljesítmény metrikák gyűjtése** ```python import time from dataclasses import dataclass from typing import Dict, List @dataclass class ToolMetrics: call_count: int = 0 total_duration: float = 0.0 error_count: int = 0 last_called: float = 0.0 class MetricsCollector: def __init__(self): self.metrics: Dict[str, ToolMetrics] = {} async def track_tool_call(self, tool_name: str, func): start_time = time.time() if tool_name not in self.metrics: self.metrics[tool_name] = ToolMetrics() metric = self.metrics[tool_name] metric.call_count += 1 metric.last_called = start_time try: result = await func() duration = time.time() - start_time metric.total_duration += duration logger.info(f"Eszköz {tool_name} - Időtartam: {duration:.2f}s") return result except Exception as e: metric.error_count += 1 logger.error(f"Eszköz {tool_name} hiba: {e}") raise def get_performance_report(self) -> str: report = "=== MCP Teljesítmény Jelentés ===\n" for tool_name, metric in self.metrics.items(): avg_duration = metric.total_duration / max(metric.call_count, 1) error_rate = metric.error_count / max(metric.call_count, 1) * 100 report += f""" Eszköz: {tool_name} - Hívások száma: {metric.call_count} - Átlagos időtartam: {avg_duration:.3f}s - Hibaarány: {error_rate:.1f}% - Legutóbbi hívás: {time.ctime(metric.last_called)} """ return report # Globális metrics collector metrics = MetricsCollector() ``` ### Biztonsági best practice-ek **Input validáció és sanitizáció** ```python from marshmallow import Schema, fields, ValidationError import re class GitHubToolSchema(Schema): repo = fields.Str(required=True, validate=lambda x: bool(re.match(r'^[\w\.-]+/[\w\.-]+, x))) branch = fields.Str(missing='main', validate=lambda x: bool(re.match(r'^[\w\.-/]+, x))) action = fields.Str(required=True, validate=lambda x: x in ['read', 'write', 'delete']) @app.call_tool() async def call_github_tool(name: str, arguments: dict): schema = GitHubToolSchema() try: validated_args = schema.load(arguments) except ValidationError as e: logger.error(f"Validációs hiba: {e.messages}") raise ValueError(f"Érvénytelen paraméterek: {e.messages}") # Biztonságos végrehajtás validált argumentumokkal return await execute_github_operation(validated_args) ``` **Hozzáférés-vezérlés implementáció** ```python from functools import wraps from typing import Set class PermissionManager: def __init__(self): self.user_permissions: Dict[str, Set[str]] = { "admin": {"read", "write", "delete", "admin"}, "developer": {"read", "write"}, "viewer": {"read"} } def has_permission(self, user_role: str, required_permission: str) -> bool: return required_permission in self.user_permissions.get(user_role, set()) permission_manager = PermissionManager() def require_permission(permission: str): def decorator(func): @wraps(func) async def wrapper(*args, **kwargs): # Felhasználói szerepkör lekérése (session-ből vagy token-ből) user_role = get_current_user_role() if not permission_manager.has_permission(user_role, permission): raise PermissionError(f"Nincs jogosultság '{permission}' művelethez") return await func(*args, **kwargs) return wrapper return decorator @app.call_tool() @require_permission("write") async def create_repository(name: str, arguments: dict): # Csak 'write' jogosultsággal rendelkező felhasználók hívhatják meg return await github_create_repo(arguments) ``` ### Skálázhatóság és load balancing **Redis-alapú session megosztás** ```python import redis.asyncio as redis import json from typing import Optional class RedisSessionManager: def __init__(self, redis_url: str = "redis://localhost:6379"): self.redis_pool = redis.ConnectionPool.from_url(redis_url) self.redis_client = redis.Redis(connection_pool=self.redis_pool) async def store_session(self, session_id: str, data: dict, ttl: int = 3600): """Session adatok tárolása Redis-ben""" await self.redis_client.setex( f"mcp:session:{session_id}", ttl, json.dumps(data) ) async def get_session(self, session_id: str) -> Optional[dict]: """Session adatok lekérése""" data = await self.redis_client.get(f"mcp:session:{session_id}") return json.loads(data) if data else None async def delete_session(self, session_id: str): """Session törlése""" await self.redis_client.delete(f"mcp:session:{session_id}") # Globális session manager session_manager = RedisSessionManager() @app.initialize() async def initialize(session_id: str): # Session inicializálás és tárolás session_data = { "created_at": time.time(), "user_id": get_authenticated_user_id(), "permissions": get_user_permissions() } await session_manager.store_session(session_id, session_data) logger.info(f"Session inicializálva: {session_id}") ``` **Kubernetes deployment konfiguráció** ```yaml apiVersion: apps/v1 kind: Deployment metadata: name: hungarian-mcp-server labels: app: hungarian-mcp spec: replicas: 3 selector: matchLabels: app: hungarian-mcp template: metadata: labels: app: hungarian-mcp spec: containers: - name: mcp-server image: hungarian-mcp:latest ports: - containerPort: 8080 env: - name: REDIS_URL value: "redis://redis-cluster:6379" - name: LOG_LEVEL value: "INFO" - name: METRICS_ENABLED value: "true" resources: requests: memory: "256Mi" cpu: "250m" limits: memory: "512Mi" cpu: "500m" readinessProbe: httpGet: path: /health port: 8080 initialDelaySeconds: 10 periodSeconds: 5 livenessProbe: httpGet: path: /health port: 8080 initialDelaySeconds: 30 periodSeconds: 10 --- apiVersion: v1 kind: Service metadata: name: hungarian-mcp-service spec: selector: app: hungarian-mcp ports: - port: 80 targetPort: 8080 type: LoadBalancer --- apiVersion: autoscaling/v2 kind: HorizontalPodAutoscaler metadata: name: hungarian-mcp-hpa spec: scaleTargetRef: apiVersion: apps/v1 kind: Deployment name: hungarian-mcp-server minReplicas: 2 maxReplicas: 10 metrics: - type: Resource resource: name: cpu target: type: Utilization averageUtilization: 70 - type: Resource resource: name: memory target: type: Utilization averageUtilization: 80 ``` ### Fejlett eszköz készítés minták **Dinamikus eszköz regisztráció** ```python from typing import Dict, Any, List import inspect class DynamicToolRegistry: def __init__(self, server: Server): self.server = server self.registered_tools: Dict[str, Tool] = {} def register_tool_from_function(self, func, name: str = None, description: str = None): """Python függvényből automatikusan eszköz regisztrálása""" tool_name = name or func.__name__ tool_description = description or func.__doc__ or f"Automatically generated tool for {func.__name__}" # Függvény szignatúra elemzése sig = inspect.signature(func) properties = {} required = [] for param_name, param in sig.parameters.items(): if param_name in ['self', 'cls']: continue param_type = "string" # Alapértelmezett if param.annotation == int: param_type = "integer" elif param.annotation == float: param_type = "number" elif param.annotation == bool: param_type = "boolean" properties[param_name] = {"type": param_type} if param.default == param.empty: required.append(param_name) # Eszköz schema generálása input_schema = { "type": "object", "properties": properties, "required": required } tool = Tool( name=tool_name, description=tool_description, inputSchema=input_schema ) self.registered_tools[tool_name] = tool # Eszköz handler regisztrálása @self.server.call_tool() async def tool_handler(name: str, arguments: dict): if name == tool_name: try: # Argumentumok átadása a függvénynek if inspect.iscoroutinefunction(func): result = await func(**arguments) else: result = func(**arguments) return TextContent(type="text", text=str(result)) except Exception as e: logger.error(f"Hiba az eszköz végrehajtása során {tool_name}: {e}") raise logger.info(f"Dinamikusan regisztrált eszköz: {tool_name}") return tool # Használat példa def calculate_hungarian_tax(gross_income: int, tax_year: int = 2024) -> str: """Magyar adókalkulátor - bruttó jövedelem alapján számítja ki a fizetendő adót""" # 2024-es magyar adótáblázat (egyszerűsített) tax_rates = { 2024: { "personal_allowance": 77300, # Személyi kedvezmény "tax_rate": 0.15 # SZJA alapkulcs } } if tax_year not in tax_rates: return f"Nincs adat a {tax_year} adóévhez" rates = tax_rates[tax_year] taxable_income = max(0, gross_income - rates["personal_allowance"]) tax_amount = taxable_income * rates["tax_rate"] net_income = gross_income - tax_amount return f""" Magyar adókalkuláció ({tax_year}): - Bruttó jövedelem: {gross_income:,} Ft - Személyi kedvezmény: {rates['personal_allowance']:,} Ft - Adóalap: {taxable_income:,} Ft - SZJA ({rates['tax_rate']*100}%): {tax_amount:,} Ft - Nettó jövedelem: {net_income:,} Ft """ # Dinamikus regisztráció tool_registry = DynamicToolRegistry(app) tool_registry.register_tool_from_function( calculate_hungarian_tax, description="Magyar személyijövedelemadó kalkulátor" ) ``` **Kontextus-tudatos eszközök** ```python class ContextAwareToolMixin: def __init__(self): self.context_history: List[Dict[str, Any]] = [] self.user_preferences: Dict[str, Any] = {} async def add_context(self, context: Dict[str, Any]): """Kontextus hozzáadása a történethez""" context['timestamp'] = time.time() self.context_history.append(context) # Csak az utolsó 50 kontextust tartjuk meg if len(self.context_history) > 50: self.context_history = self.context_history[-50:] def get_relevant_context(self, query: str, limit: int = 5) -> List[Dict[str, Any]]: """Releváns kontextus lekérése a query alapján""" # Egyszerű kulcsszó alapú relevanciavizsgálat query_words = set(query.lower().split()) scored_contexts = [] for ctx in self.context_history: ctx_text = str(ctx.get('content', '')) ctx_words = set(ctx_text.lower().split()) # Jaccard similarity intersection = len(query_words.intersection(ctx_words)) union = len(query_words.union(ctx_words)) if union > 0: similarity = intersection / union scored_contexts.append((similarity, ctx)) # Relevancia szerint rendezés scored_contexts.sort(key=lambda x: x[0], reverse=True) return [ctx for _, ctx in scored_contexts[:limit]] class SmartDocumentationTool(ContextAwareToolMixin): def __init__(self): super().__init__() self.project_docs: Dict[str, str] = {} async def generate_documentation(self, project_name: str, code_context: str, doc_type: str = "API") -> str: """Intelligens dokumentáció generálás kontextus alapján""" # Releváns kontextus lekérése relevant_contexts = self.get_relevant_context( f"{project_name} {code_context} {doc_type}" ) # Kontextus alapján dokumentáció stílus személyreszabása style_preferences = self.user_preferences.get('doc_style', 'detailed') language = self.user_preferences.get('language', 'hungarian') doc_template = self._get_doc_template(doc_type, style_preferences, language) # Kontextus integráció context_summary = self._summarize_context(relevant_contexts) documentation = doc_template.format( project_name=project_name, code_context=code_context, context_summary=context_summary, timestamp=time.strftime('%Y-%m-%d %H:%M:%S') ) # Kontextus frissítése await self.add_context({ 'type': 'documentation_generated', 'project': project_name, 'doc_type': doc_type, 'content': documentation[:500] # Csak az első 500 karakter tárolása }) return documentation def _get_doc_template(self, doc_type: str, style: str, language: str) -> str: templates = { 'API': { 'hungarian': { 'brief': "# {project_name} API Dokumentáció\n\n{code_context}\n\nGenerálva: {timestamp}", 'detailed': """# {project_name} API Részletes Dokumentáció ## Áttekintés {code_context} ## Kontextus alapú javaslatok {context_summary} ## Használati példák [Automatikusan generált példák következnek...] --- Dokumentáció generálva: {timestamp} Korábbi kontextus alapján személyre szabva. """ } } } return templates.get(doc_type, {}).get(language, {}).get(style, "# {project_name}\n{code_context}\n\nGenerated: {timestamp}") def _summarize_context(self, contexts: List[Dict[str, Any]]) -> str: if not contexts: return "Nincs korábbi kontextus elérhető." summary_parts = [] for ctx in contexts: ctx_type = ctx.get('type', 'unknown') content_preview = str(ctx.get('content', ''))[:100] summary_parts.append(f"- {ctx_type}: {content_preview}...") return "Korábbi munkák alapján:\n" + "\n".join(summary_parts) # Regisztráció smart_doc_tool = SmartDocumentationTool() @app.call_tool() async def smart_documentation(name: str, arguments: dict): if name == "generate_smart_docs": return TextContent( type="text", text=await smart_doc_tool.generate_documentation( arguments.get('project_name'), arguments.get('code_context'), arguments.get('doc_type', 'API') ) ) ``` Ez a teljes magyar nyelvű MCP dokumentáció gyakorlati implementációs útmutatókkal, monitoring technikákkal, biztonsági best practice-ekkel és fejlett eszközkészítési mintákkal egészíti ki az eredeti tartalmat.