feat: enhance MCP server with SSE support and Docker port exposure

This commit is contained in:
rodolpho
2026-05-07 21:49:10 -03:00
parent 3e4e871144
commit 6526193cbe
2 changed files with 29 additions and 13 deletions
+4 -2
View File
@@ -23,8 +23,10 @@ RUN pip install --no-cache-dir -r requirements.txt
# Copiar o restante do código # Copiar o restante do código
COPY . . COPY . .
# Expor a porta do MkDocs # Expor as portas do MkDocs e do MCP (SSE)
EXPOSE 8000 EXPOSE 8000
EXPOSE 8001
# Comando padrão: rodar o servidor da Wiki # Comando padrão: agora roda um pequeno entrypoint que sobe os dois serviços se desejado
# Por padrão, sobe a Wiki.
CMD ["mkdocs", "serve", "-a", "0.0.0.0:8000"] CMD ["mkdocs", "serve", "-a", "0.0.0.0:8000"]
+25 -11
View File
@@ -1,5 +1,6 @@
import json import json
import os import os
import argparse
from typing import List, Dict from typing import List, Dict
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
@@ -19,32 +20,35 @@ def load_chunks() -> List[Dict]:
@mcp.tool() @mcp.tool()
def search_docs(query: str) -> str: def search_docs(query: str) -> str:
""" """
Busca informações técnicas na documentação do Fluig por palavras-chave. Busca profunda na documentação técnica do Fluig.
Retorna os chunks mais relevantes. Retorna trechos exatos sobre APIs, Eventos, Datasets e Configurações.
""" """
chunks = load_chunks() chunks = load_chunks()
results = [] results = []
query = query.lower() query = query.lower()
# Busca com prioridade em títulos
for chunk in chunks: for chunk in chunks:
content = chunk.get("content", "").lower()
title = chunk.get("metadata", {}).get("title", "").lower() title = chunk.get("metadata", {}).get("title", "").lower()
content = chunk.get("content", "").lower()
if query in content or query in title: if query in title:
results.insert(0, f"### [ALTA RELEVÂNCIA] {chunk['metadata'].get('title')}\n{chunk['content']}\n---\n")
elif query in content:
results.append(f"### {chunk['metadata'].get('title')}\n{chunk['content']}\n---\n") results.append(f"### {chunk['metadata'].get('title')}\n{chunk['content']}\n---\n")
if len(results) >= 5: # Limita a 5 resultados para não estourar contexto if len(results) >= 8: # Limite expandido para busca mais rica
break break
if not results: if not results:
return f"Nenhuma informação encontrada para: {query}" return f"Nenhuma informação técnica encontrada para: {query}. Tente termos como 'createDataset', 'hAPI', 'OAuth' ou 'BPM'."
return "\n".join(results) return "\n".join(results)
@mcp.tool() @mcp.tool()
def get_code_snippets(language: str) -> str: def get_code_snippets(language: str) -> str:
""" """
Recupera exemplos de código (snippets) para uma linguagem específica (javascript, java ou sql). Recupera a biblioteca completa de exemplos de código para Fluig (javascript, java ou sql).
""" """
lang = language.lower() lang = language.lower()
file_map = { file_map = {
@@ -63,15 +67,25 @@ def get_code_snippets(language: str) -> str:
with open(path, "r", encoding="utf-8") as f: with open(path, "r", encoding="utf-8") as f:
return f.read() return f.read()
return f"Arquivo de snippets para {language} não encontrado." return f"Arquivo de snippets para {language} não encontrado. Certifique-se de que o snippet_generator.py foi executado."
@mcp.resource("docs://all_titles") @mcp.resource("docs://all_titles")
def list_all_titles() -> str: def list_all_titles() -> str:
"""Lista todos os títulos de documentos disponíveis na wiki.""" """Lista todos os temas técnicos disponíveis na base de conhecimento."""
chunks = load_chunks() chunks = load_chunks()
titles = sorted(list(set(chunk["metadata"].get("title") for chunk in chunks))) titles = sorted(list(set(chunk["metadata"].get("title") for chunk in chunks)))
return "\n".join(titles) return "\n".join(titles)
if __name__ == "__main__": if __name__ == "__main__":
# Rodar o servidor (padrão usa STDIO para integração com Claude/IDEs) parser = argparse.ArgumentParser(description="Fluig MCP Server")
mcp.run() parser.add_argument("--mode", choices=["stdio", "sse"], default="stdio", help="Modo de operação (padrão: stdio)")
parser.add_argument("--port", type=int, default=8001, help="Porta para o modo SSE (padrão: 8001)")
args = parser.parse_args()
if args.mode == "sse":
print(f"Iniciando servidor MCP em modo SSE na porta {args.port}...")
mcp.run(transport="sse")
else:
# Modo STDIO (padrão para Claude Desktop e integração local)
mcp.run(transport="stdio")