# -*- coding: utf-8 -*- # app.py import sys import os import asyncio from functools import wraps from flask import Flask, request, jsonify, Response, redirect from flask_cors import CORS from flasgger import Swagger from werkzeug.exceptions import BadRequest from config import settings # Bibliotecas do MCP e LangChain Puro from mcp.client.stdio import stdio_client, StdioServerParameters from mcp.client.session import ClientSession from langchain_openai import ChatOpenAI from langchain_mcp_adapters.tools import load_mcp_tools from langchain_core.messages import HumanMessage, SystemMessage #### LOGGING #### import logging LOG_LEVEL = logging.DEBUG if settings.DEBUG else logging.INFO logging.basicConfig( level=LOG_LEVEL, format='%(asctime)s - %(levelname)s - %(message)s', stream=sys.stderr ) logger = logging.getLogger("GATEWAY_ZEMPO") # --- Inicialização do Flask --- app = Flask(__name__) CORS(app) # --- Configuração do Swagger --- template = { "swagger": "2.0", "info": { "title": "Gateway IA - Zempo Oficial", "description": "API REST para execução isolada das ferramentas de IA integradas ao Zempo. Envie seu token do Zempo no botão Authorize.", "version": "1.0.0" }, "securityDefinitions": { "Bearer": { "type": "apiKey", "name": "Authorization", "in": "header", "description": "Insira o token oficial do Zempo no formato: **Bearer SEU_TOKEN_AQUI**" } }, "security": [{"Bearer": []}] } swagger = Swagger(app, template=template) # --- Decorators de Segurança --- @app.before_request def check_docs_auth(): if request.path.startswith('/apidocs'): auth = request.authorization if not auth or not (auth.username == settings.DOCS_USERNAME and auth.password == settings.DOCS_PASSWORD): return Response('Acesso restrito à documentação.', 401, {'WWW-Authenticate': 'Basic realm="Login Required"'}) def extrair_token_zempo(f): """ Extrai o Token do Zempo enviado pelo usuário (via Chatbot ou Swagger) através do cabeçalho Authorization: Bearer . """ @wraps(f) def decorated(*args, **kwargs): auth_header = request.headers.get('Authorization') if not auth_header: return jsonify({"error": "Acesso negado. Envie o token do Zempo no cabeçalho Authorization."}), 401 try: # Extrai apenas a parte do token, ignorando a palavra "Bearer" token_usuario = auth_header.split(" ")[1] if not token_usuario: raise IndexError # Salva o token temporariamente na requisição atual request.zempo_token = token_usuario except IndexError: return jsonify({"error": "Token mal formatado. Utilize o formato: Bearer SEU_TOKEN_AQUI"}), 401 return f(*args, **kwargs) return decorated # --- Lógica da Inteligência Artificial --- async def _executar_ia_simples(pergunta: str, token_zempo_do_usuario: str): caminho_mcp = os.path.join(os.path.dirname(__file__), "mcp_server.py") # INJEÇÃO DINÂMICA: Criamos uma cópia do ambiente do servidor e adicionamos o token do usuário! env_subprocesso = os.environ.copy() env_subprocesso["USER_ZEMPO_TOKEN"] = token_zempo_do_usuario server_params = StdioServerParameters( command=sys.executable, args=[caminho_mcp], env=env_subprocesso # O mcp_server.py só lerá o token que veio desta requisição ) try: async with stdio_client(server_params) as (read_stream, write_stream): async with ClientSession(read_stream, write_stream) as session: await session.initialize() mcp_tools = await load_mcp_tools(session) if not settings.OPENAI_API_KEY: raise ValueError("Chave da OpenAI (OPENAI_API_KEY) ausente no sistema.") llm = ChatOpenAI( model="gpt-4o", temperature=0, api_key=settings.OPENAI_API_KEY ) llm_com_ferramentas = llm.bind_tools(mcp_tools) mensagens = [] contexto_zempo = ( "Você é um Assistente de Inteligência Esportiva integrado ao Zempo (sistema oficial da Confederação Brasileira de Judô). " "Seu objetivo é extrair dados usando as ferramentas fornecidas e responder ao usuário de forma clara. " "A API do Zempo é estritamente de leitura. " "Se houver algum erro de 'Token inválido' nas ferramentas, avise ao usuário que o token fornecido expirou ou é incorreto." ) mensagens.append(SystemMessage(content=contexto_zempo)) mensagens.append(HumanMessage(content=pergunta)) logger.debug(f"🤖 [IA] Analisando a intenção: '{pergunta}'") resposta_ia = await llm_com_ferramentas.ainvoke(mensagens) if resposta_ia.tool_calls: mensagens.append(resposta_ia) tools_por_nome = {t.name: t for t in mcp_tools} for tool_call in resposta_ia.tool_calls: nome_tool = tool_call["name"] args_tool = tool_call.get("args", {}) logger.debug(f"🛠️ [IA -> MCP] Ferramenta: {nome_tool} | Args: {args_tool}") ferramenta = tools_por_nome[nome_tool] resultado_ferramenta = await ferramenta.ainvoke(tool_call) logger.debug(f"🧠 [MCP -> IA] Dados recebidos: {resultado_ferramenta.content[:400]}...") mensagens.append(resultado_ferramenta) logger.debug("✍️ [IA] Compilando resposta em linguagem natural...") resposta_final = await llm_com_ferramentas.ainvoke(mensagens) return resposta_final.content logger.debug("💡 [IA] Resposta direta (sem necessidade de ferramentas).") return resposta_ia.content except Exception as e: logger.error(f"❌ [ERRO CRÍTICO] Falha na execução da IA: {e}") return f"Erro interno na IA: {str(e)}" # --- Rotas da API --- @app.route('/') def index(): return redirect('/apidocs') @app.route('/health', methods=['GET']) def health_check(): return jsonify({"status": "online", "service": "Gateway IA Zempo (CBJ)"}), 200 @app.route('/chat', methods=['POST']) @extrair_token_zempo def chat_endpoint(): """ Processa uma pergunta utilizando as ferramentas do Zempo. --- tags: - IA Esportiva Zempo security: - Bearer: [] parameters: - name: body in: body required: true schema: type: object required: - pergunta properties: pergunta: type: string example: "Quais os atletas inscritos na competição jj23?" responses: 200: description: Resposta processada pela IA. 401: description: Token da API não fornecido. """ try: data = request.get_json() if not data or 'pergunta' not in data: return jsonify({"status": "erro", "mensagem": "Campo 'pergunta' é obrigatório."}), 400 pergunta = data['pergunta'] # Recupera o token que o decorator salvou! token_do_usuario = request.zempo_token logger.debug(f"\n{'='*50}\n🚀 NOVA REQUISIÇÃO AO ASSISTENTE ZEMPO\n{'='*50}") # Inicia a corrente LangChain passando o token resposta = asyncio.run(_executar_ia_simples(pergunta, token_do_usuario)) logger.debug(f"🏁 RESPOSTA FINAL ENVIADA\n{'='*50}\n") return jsonify({"status": "sucesso", "resposta": resposta}) except BadRequest: return jsonify({"status": "erro", "mensagem": "JSON inválido na requisição."}), 400 except Exception as e: logger.error(f"Erro no endpoint chat: {e}") return jsonify({"status": "erro", "mensagem": str(e)}), 500 if __name__ == '__main__': print("✅ APP.PY GATEWAY ZEMPO Carregado.") if settings.DEBUG: app.run(debug=True, port=5001) else: app.run(debug=False)