João Henrique
b541f502ba
- Adicionado estrutura completa do projeto - Configurado MCP server para Premiere Pro - Adicionado documentação e skills - Configurado Gitignore para o projeto
72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
from abc import ABC, abstractmethod
|
|
import json
|
|
import subprocess
|
|
from typing import Any
|
|
|
|
|
|
class ClienteMCP(ABC):
|
|
"""Define a comunicação técnica com o servidor MCP."""
|
|
|
|
@abstractmethod
|
|
def conectar(self) -> None: ...
|
|
|
|
@abstractmethod
|
|
def desconectar(self) -> None: ...
|
|
|
|
@abstractmethod
|
|
def esta_conectado(self) -> bool: ...
|
|
|
|
@abstractmethod
|
|
def chamar(self, nome_da_ferramenta: str, argumentos: dict[str, Any] | None = None) -> dict[str, Any]: ...
|
|
|
|
|
|
class ClienteMCPPorStdio(ClienteMCP):
|
|
"""Cliente MCP para o servidor local do Premiere via stdio."""
|
|
|
|
def __init__(self, comando: list[str], tempo_limite: float = 30.0) -> None:
|
|
self.comando = comando
|
|
self.tempo_limite = tempo_limite
|
|
self._processo: subprocess.Popen[str] | None = None
|
|
self._proximo_id = 1
|
|
|
|
def conectar(self) -> None:
|
|
if self._processo is not None:
|
|
return
|
|
self._processo = subprocess.Popen(
|
|
self.comando,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
bufsize=1,
|
|
)
|
|
self._enviar("initialize", {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "engine", "version": "0.1.0"}})
|
|
|
|
def desconectar(self) -> None:
|
|
if self._processo is not None:
|
|
self._processo.terminate()
|
|
self._processo = None
|
|
|
|
def esta_conectado(self) -> bool:
|
|
return self._processo is not None and self._processo.poll() is None
|
|
|
|
def chamar(self, nome_da_ferramenta: str, argumentos: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
if not self.esta_conectado():
|
|
raise RuntimeError("Cliente MCP não está conectado.")
|
|
resposta = self._enviar("tools/call", {"name": nome_da_ferramenta, "arguments": argumentos or {}})
|
|
if resposta.get("error"):
|
|
raise RuntimeError(f"Erro MCP: {resposta['error']}")
|
|
return resposta.get("result", {})
|
|
|
|
def _enviar(self, metodo: str, parametros: dict[str, Any]) -> dict[str, Any]:
|
|
if self._processo is None or self._processo.stdin is None or self._processo.stdout is None:
|
|
raise RuntimeError("Processo MCP indisponível.")
|
|
identificador = self._proximo_id
|
|
self._proximo_id += 1
|
|
self._processo.stdin.write(json.dumps({"jsonrpc": "2.0", "id": identificador, "method": metodo, "params": parametros}) + "\n")
|
|
self._processo.stdin.flush()
|
|
linha = self._processo.stdout.readline()
|
|
if not linha:
|
|
raise RuntimeError("O MCP encerrou sem retornar resposta.")
|
|
return json.loads(linha)
|