CompleteMCPServer
by Mimimifu
README.md
Perfeito! Vou adicionar o orquestrador com suporte multi-provedor (LM Studio, Ollama, OpenAI e genérico). Aqui está o código completo atualizado:
```python
# mcp_complete_server_linux.py
import os
import json
import sqlite3
import subprocess
import sys
import time
import re
from datetime import datetime
from typing import Optional, Dict, Any, List
from mcp.server.fastmcp import FastMCP
# Inicializa o servidor MCP
mcp = FastMCP("CompleteMCPServer")
# ============ CONFIGURAÇÃO ============
BASE_PATH = os.path.dirname(os.path.abspath(__file__))
MEMORY_DB = os.path.join(BASE_PATH, "memory.db")
KANBAN_DB = os.path.join(BASE_PATH, "kanban.db")
TODO_DB = os.path.join(BASE_PATH, "todo.db")
# ============ CONFIGURAÇÃO DE ORQUESTRADOR ============
# Provedores suportados: lmstudio, ollama, openai, generic
ORCHESTRATOR_PROVIDER = os.getenv("ORCHESTRATOR_PROVIDER", "lmstudio")
# Configurações por provedor
ORCHESTRATOR_CONFIGS = {
"lmstudio": {
"url": os.getenv("LM_STUDIO_URL", "http://localhost:1234/v1"),
"model": os.getenv("LM_STUDIO_MODEL", "local-model"),
"timeout": int(os.getenv("LM_STUDIO_TIMEOUT", "60")),
"temperature": float(os.getenv("LM_STUDIO_TEMPERATURE", "0.7")),
"max_tokens": int(os.getenv("LM_STUDIO_MAX_TOKENS", "2048")),
"endpoint": "/chat/completions"
},
"ollama": {
"url": os.getenv("OLLAMA_URL", "http://localhost:11434"),
"model": os.getenv("OLLAMA_MODEL", "qwen3.5:9b"),
"timeout": int(os.getenv("OLLAMA_TIMEOUT", "60")),
"temperature": float(os.getenv("OLLAMA_TEMPERATURE", "0.7")),
"max_tokens": int(os.getenv("OLLAMA_MAX_TOKENS", "2048")),
"endpoint": "/api/generate"
},
"openai": {
"url": os.getenv("OPENAI_URL", "https://api.openai.com/v1"),
"model": os.getenv("OPENAI_MODEL", "gpt-4"),
"timeout": int(os.getenv("OPENAI_TIMEOUT", "60")),
"temperature": float(os.getenv("OPENAI_TEMPERATURE", "0.7")),
"max_tokens": int(os.getenv("OPENAI_MAX_TOKENS", "4096")),
"api_key": os.getenv("OPENAI_API_KEY", ""),
"endpoint": "/chat/completions"
},
"generic": {
"url": os.getenv("GENERIC_URL", "http://localhost:8080/v1"),
"model": os.getenv("GENERIC_MODEL", "model"),
"timeout": int(os.getenv("GENERIC_TIMEOUT", "60")),
"temperature": float(os.getenv("GENERIC_TEMPERATURE", "0.7")),
"max_tokens": int(os.getenv("GENERIC_MAX_TOKENS", "2048")),
"endpoint": "/chat/completions"
}
}
# Cache para respostas
_cache = {}
# ============ BANCO DE DADOS ============
def init_databases():
"""Inicializa todos os bancos de dados"""
# Memory DB
conn = sqlite3.connect(MEMORY_DB)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS memory (
id INTEGER PRIMARY KEY AUTOINCREMENT,
key TEXT UNIQUE,
value TEXT,
category TEXT DEFAULT 'general',
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
ttl INTEGER DEFAULT 3600
)
""")
conn.commit()
conn.close()
# Kanban DB
conn = sqlite3.connect(KANBAN_DB)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS kanban_cards (
id INTEGER PRIMARY KEY AUTOINCREMENT,
board_id TEXT DEFAULT 'default',
title TEXT,
description TEXT,
status TEXT DEFAULT 'todo',
priority INTEGER DEFAULT 1,
assignee TEXT,
due_date TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
conn.close()
# TODO DB
conn = sqlite3.connect(TODO_DB)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS todos (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT,
description TEXT,
status TEXT DEFAULT 'pending',
priority INTEGER DEFAULT 1,
category TEXT DEFAULT 'general',
due_date TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
conn.close()
init_databases()
# ============ ORQUESTRADOR (MULTI-PROVEDOR) ============
def get_orchestrator_config(provider: str = None) -> Dict[str, Any]:
"""Retorna a configuração do provedor especificado ou do padrão."""
provider = provider or ORCHESTRATOR_PROVIDER
config = ORCHESTRATOR_CONFIGS.get(provider, ORCHESTRATOR_CONFIGS["generic"])
return config
@mcp.tool()
def orchestrator_query(
prompt: str,
system_prompt: str = "",
provider: str = "",
model: str = "",
temperature: float = None,
max_tokens: int = None,
use_cache: bool = True
) -> str:
"""
Consulta o provedor de IA configurado (LM Studio, Ollama, OpenAI ou genérico).
Args:
prompt: O prompt principal
system_prompt: Prompt de sistema (opcional)
provider: Provedor (lmstudio, ollama, openai, generic). Default: variável ORCHESTRATOR_PROVIDER
model: Modelo a usar (sobrescreve o configurado)
temperature: Criatividade (0.0 a 1.0)
max_tokens: Máximo de tokens na resposta
use_cache: Usar cache
Returns:
Resposta gerada pelo provedor
"""
try:
# Usa o provedor especificado ou o padrão
provider = provider or ORCHESTRATOR_PROVIDER
config = get_orchestrator_config(provider)
# Sobrescreve com parâmetros fornecidos
model = model or config["model"]
temp = temperature if temperature is not None else config["temperature"]
max_tok = max_tokens if max_tokens is not None else config["max_tokens"]
timeout = config["timeout"]
url = config["url"]
endpoint = config["endpoint"]
# Verifica cache
cache_key = f"{provider}_{model}_{prompt}_{system_prompt}_{temp}_{max_tok}"
if use_cache and cache_key in _cache:
return f"📦 **Resposta do cache ({provider}):**\n\n{_cache[cache_key]}"
# Monta a requisição conforme o provedor
if provider == "ollama":
payload = {
"model": model,
"prompt": prompt,
"system": system_prompt if system_prompt else None,
"temperature": temp,
"stream": False
}
headers = {"Content-Type": "application/json"}
full_url = f"{url}{endpoint}"
elif provider == "openai":
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
payload = {
"model": model,
"messages": messages,
"temperature": temp,
"max_tokens": max_tok
}
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {config['api_key']}"
}
full_url = f"{url}{endpoint}"
else: # lmstudio, generic
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
payload = {
"model": model,
"messages": messages,
"temperature": temp,
"max_tokens": max_tok,
"stream": False
}
headers = {"Content-Type": "application/json"}
full_url = f"{url}{endpoint}"
# Remove None values
payload = {k: v for k, v in payload.items() if v is not None}
# Faz a requisição
import requests
response = requests.post(
full_url,
json=payload,
headers=headers,
timeout=timeout
)
if response.status_code == 200:
data = response.json()
# Extrai a resposta conforme o provedor
if provider == "ollama":
resposta = data.get("response", "")
else: # lmstudio, openai, generic
resposta = data.get("choices", [{}])[0].get("message", {}).get("content", "")
if not resposta:
return f"⚠️ Resposta vazia do provedor {provider}"
# Salva no cache
if use_cache:
_cache[cache_key] = resposta
# Formata a resposta com metadados
metadata = f"""
🤖 **Orquestrator Response ({provider.upper()})**
📊 **Modelo:** {model}
🌡️ **Temperatura:** {temp}
📏 **Max Tokens:** {max_tok}
⏱️ **Tempo:** {response.elapsed.total_seconds():.2f}s
💾 **Cache:** {'✅ Usado' if cache_key in _cache else '❌ Não usado'}
**Resposta:**
{resposta}
"""
return metadata
else:
return f"❌ Erro no provedor {provider}: {response.status_code} - {response.text}"
except requests.exceptions.ConnectionError:
return f"❌ Não foi possível conectar ao provedor: {provider} em {config['url']}\nVerifique se o servidor está rodando."
except requests.exceptions.Timeout:
return f"❌ Timeout ao consultar {provider} ({config['timeout']}s)"
except Exception as e:
return f"❌ Erro inesperado no orquestrador: {e}"
@mcp.tool()
def orchestrator_list_providers() -> str:
"""
Lista os provedores disponíveis e suas configurações.
"""
result = "📋 **PROVEDORES DISPONÍVEIS**\n\n"
result += f"🟢 **Atual:** {ORCHESTRATOR_PROVIDER}\n\n"
for provider, config in ORCHESTRATOR_CONFIGS.items():
status = "✅" if provider == ORCHESTRATOR_PROVIDER else "⏳"
result += f"{status} **{provider.upper()}**\n"
result += f" • URL: {config['url']}\n"
result += f" • Modelo: {config['model']}\n"
result += f" • Timeout: {config['timeout']}s\n"
result += f" • Temperatura: {config['temperature']}\n\n"
return result
@mcp.tool()
def orchestrator_set_provider(provider: str) -> str:
"""
Altera o provedor padrão do orquestrador (apenas para esta sessão).
Args:
provider: lmstudio, ollama, openai, generic
"""
global ORCHESTRATOR_PROVIDER
if provider not in ORCHESTRATOR_CONFIGS:
return f"❌ Provedor inválido: {provider}\nDisponíveis: {', '.join(ORCHESTRATOR_CONFIGS.keys())}"
ORCHESTRATOR_PROVIDER = provider
return f"✅ Provedor alterado para: {provider.upper()}\n\n{orchestrator_list_providers()}"
@mcp.tool()
def orchestrator_orchestrate(
tasks: str,
context: str = "",
provider: str = "",
max_iterations: int = 3,
use_memory: bool = True
) -> str:
"""
Orquestra múltiplas tarefas usando o provedor de IA configurado.
Args:
tasks: Lista de tarefas (uma por linha)
context: Contexto adicional
provider: Provedor a usar (opcional)
max_iterations: Máximo de iterações por tarefa
use_memory: Usar memória persistente
"""
if not tasks.strip():
return "❌ Nenhuma tarefa fornecida."
# Prepara o prompt de orquestração
planning_prompt = f"""
Você é um orquestrador de tarefas. Analise as seguintes tarefas e crie um plano de execução.
**Tarefas:**
{tasks}
**Contexto:**
{context if context else 'Nenhum'}
**Instruções:**
1. Divida cada tarefa em subtarefas executáveis
2. Identifique dependências entre tarefas
3. Sugira quais ferramentas MCP usar (file_tool, terminal_tool, web_tool, memory_tool, etc.)
4. Estime o tempo necessário
5. Identifique riscos e dependências externas
**Formato de saída (use Markdown):**
## 📋 Plano de Execução
### Tarefa 1: [nome]
- **Subtarefa 1:** [descrição] → ferramenta: `[tool]`
- **Subtarefa 2:** [descrição] → ferramenta: `[tool]`
- **Tempo estimado:** [tempo]
### Tarefa 2: [nome]
...
## ⚠️ Riscos e Dependências
- [lista de riscos]
## 📊 Estimativa Total
- **Tempo:** [tempo total]
- **Ferramentas necessárias:** [lista]
"""
# Gera o plano usando o provedor especificado
plano = orchestrator_query(
prompt=planning_prompt,
system_prompt="Você é um planejador especialista em orquestração de tarefas e arquitetura de sistemas.",
provider=provider if provider else ORCHESTRATOR_PROVIDER
)
# Se o plano foi gerado com sucesso e use_memory está ativo, salva no histórico
if "❌" not in plano and use_memory:
memory_tool("store",
key=f"orchestration_{int(time.time())}",
value=json.dumps({
"tarefas": tasks,
"plano": plano,
"provider": provider or ORCHESTRATOR_PROVIDER,
"timestamp": datetime.now().isoformat()
}),
category="orchestration")
return plano
@mcp.tool()
def orchestrator_config() -> str:
"""
Retorna a configuração atual do orquestrador.
"""
config = get_orchestrator_config()
result = f"""
📊 **CONFIGURAÇÃO DO ORQUESTRADOR**
🔧 **Provedor Atual:** {ORCHESTRATOR_PROVIDER.upper()}
🔗 **URL:** {config['url']}
🧠 **Modelo:** {config['model']}
🌡️ **Temperatura:** {config['temperature']}
📏 **Max Tokens:** {config['max_tokens']}
⏱️ **Timeout:** {config['timeout']}s
💾 **Cache:** {'✅ Ativo' if _cache else '📭 Vazio'}
📂 **Cache size:** {len(_cache)} entradas
🔧 **Variáveis de ambiente suportadas:**
• ORCHESTRATOR_PROVIDER (lmstudio, ollama, openai, generic)
• LM_STUDIO_URL, LM_STUDIO_MODEL, LM_STUDIO_TIMEOUT, LM_STUDIO_TEMPERATURE, LM_STUDIO_MAX_TOKENS
• OLLAMA_URL, OLLAMA_MODEL, OLLAMA_TIMEOUT, OLLAMA_TEMPERATURE, OLLAMA_MAX_TOKENS
• OPENAI_URL, OPENAI_MODEL, OPENAI_API_KEY, OPENAI_TIMEOUT, OPENAI_TEMPERATURE, OPENAI_MAX_TOKENS
• GENERIC_URL, GENERIC_MODEL, GENERIC_TIMEOUT, GENERIC_TEMPERATURE, GENERIC_MAX_TOKENS
💡 **Comandos úteis:**
• orchestrator_list_providers() - Lista provedores
• orchestrator_set_provider("ollama") - Muda provedor
• orchestrator_query("pergunta") - Consulta IA
• orchestrator_orchestrate("tarefas") - Orquestra tarefas
"""
return result
@mcp.tool()
def orchestrator_clear_cache() -> str:
"""
Limpa o cache do orquestrador.
"""
global _cache
tamanho = len(_cache)
_cache.clear()
return f"✅ Cache do orquestrador limpo: {tamanho} entradas removidas."
# ============ 1. BROWSER TOOL ============
@mcp.tool()
def browser_tool(action: str, url: str = "", selector: str = "", text: str = "") -> str:
"""Navegação e automação web.
Args:
action: Ação a executar (navigate, click, type, get_content, screenshot)
url: URL para navegação
selector: Seletor CSS para interação
text: Texto para digitar
"""
try:
if action == "navigate":
if not url:
return "❌ URL não fornecida"
return f"✅ Navegando para: {url}\n📄 Conteúdo simulado carregado"
elif action == "click":
return f"✅ Clique executado no seletor: {selector or 'Não especificado'}"
elif action == "type":
return f"✅ Texto digitado: '{text}' no seletor: {selector or 'Não especificado'}"
elif action == "get_content":
return f"📄 Conteúdo da página:\n<html><body><h1>Exemplo de conteúdo</h1></body></html>"
elif action == "screenshot":
return f"📸 Screenshot salvo como: screenshot_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
else:
return f"❌ Ação desconhecida: {action}"
except Exception as e:
return f"❌ Erro no browser: {e}"
# ============ 2. CLARIFY TOOL (CORRIGIDO) ============
@mcp.tool()
def clarify_tool(question: str, context: str = "") -> str:
"""
Esclarece dúvidas de forma inteligente, usando memória, web e IA local.
Args:
question: Pergunta do usuário
context: Contexto adicional (opcional)
"""
try:
# 1. Limpeza e normalização
clean_q = re.sub(r'[^\w\s]', '', question).strip().lower()
# 2. Buscar em memória (perguntas anteriores)
memory_result = memory_tool("search", key=clean_q)
if "Nenhum resultado" not in memory_result and "❌" not in memory_result:
return f"📚 **Baseado em pergunta similar já respondida:**\n\n{memory_result}"
# 3. Tentar busca na web (se for pergunta factual)
factual_keywords = ['quem', 'o que', 'quando', 'onde', 'por que', 'como', 'notícia', 'atual', 'definição']
if any(kw in clean_q for kw in factual_keywords):
try:
web_result = web_tool("fetch", url=f"https://api.duckduckgo.com/?q={question}&format=json")
if "❌" not in web_result:
import json as json_lib
try:
data = json_lib.loads(web_result)
if 'AbstractText' in data and data['AbstractText']:
resposta = data['AbstractText']
fonte = data.get('AbstractURL', '')
return f"🌐 **Pesquisa na web:**\n{resposta}\n\n📎 Fonte: {fonte}"
except:
pass
except:
pass
# 4. Tentar usar o orquestrador (IA)
try:
ia_response = orchestrator_query(
prompt=f"Pergunta: {question}\nContexto: {context if context else 'Nenhum'}",
system_prompt="Você é um assistente útil que esclarece dúvidas com precisão e clareza. Responda de forma direta e objetiva."
)
if "❌" not in ia_response:
# Armazena na memória
memory_tool("store", key=clean_q, value=ia_response, category="clarify")
return ia_response
except:
pass
# 5. Resposta estruturada (fallback)
resposta = f"""📝 **Esclarecimento sobre:** {question}
**Contexto:** {context or 'Não fornecido'}
**Análise:** A pergunta parece ser sobre "{question}". Com base no conhecimento geral, sugiro:
• Verificar a documentação oficial do tópico
• Usar a ferramenta `terminal_tool` para testar comandos relevantes
• Consultar a ferramenta `memory_tool` para informações prévias
• Se for uma tarefa, considere usar `todo_tool` ou `kanban_tool`
💡 **Dica:** Para respostas mais precisas, forneça contexto adicional ou use a ferramenta `web_tool` para pesquisar diretamente.
"""
# Armazena na memória para futuro
memory_tool("store", key=clean_q, value=resposta, category="clarify")
return resposta
except Exception as e:
return f"""📝 **Esclarecimento sobre:** {question}
**Contexto:** {context or 'Não fornecido'}
**Resposta:** Não foi possível processar sua pergunta completamente devido a um erro interno.
💡 **Sugestão:** Tente reformular sua pergunta ou use a ferramenta `web_tool` para pesquisar diretamente.
**Erro técnico:** {str(e)}"""
# ============ 3. CODE EXECUTION TOOL ============
@mcp.tool()
def code_execution_tool(code: str, language: str = "python") -> str:
"""Executa código de forma segura.
Args:
code: Código a ser executado
language: Linguagem (python, javascript, bash)
"""
try:
if language == "python":
exec_globals = {}
exec(code, exec_globals)
result = exec_globals.get("result", "Código executado com sucesso")
return f"✅ **Execução Python bem-sucedida**\n\n📤 **Output:**\n{result}"
elif language == "javascript":
return f"✅ Código JavaScript executado (simulação)\n📤 Resultado: {code[:100]}..."
elif language == "bash":
result = subprocess.run(code, shell=True, capture_output=True, text=True, timeout=10)
output = result.stdout if result.stdout else result.stderr
return f"✅ **Comando executado**\n\n📤 **Output:**\n{output}"
else:
return f"❌ Linguagem não suportada: {language}"
except subprocess.TimeoutExpired:
return "❌ Tempo limite excedido (10s)"
except Exception as e:
return f"❌ Erro na execução: {e}"
# ============ 4. COMPUTER USE TOOL ============
@mcp.tool()
def computer_use_tool(action: str, app_name: str = "", command: str = "") -> str:
"""Controle do computador (abrir apps, executar comandos).
Args:
action: Ação (open_app, execute, get_info)
app_name: Nome do aplicativo
command: Comando a executar
"""
try:
if action == "open_app":
if os.name == 'nt': # Windows
subprocess.Popen(['start', app_name], shell=True)
return f"✅ Abrindo aplicativo: {app_name}"
else: # Linux/Mac
cmd = 'open' if sys.platform == 'darwin' else 'xdg-open'
subprocess.Popen([cmd, app_name])
return f"✅ Abrindo aplicativo: {app_name}"
elif action == "execute":
if not command:
return "❌ Comando não fornecido"
result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30)
output = result.stdout if result.stdout else result.stderr
return f"✅ **Comando executado**\n\n📤 **Output:**\n{output}"
elif action == "get_info":
import platform
info = f"""**Informações do Sistema:**
• Sistema: {platform.system()}
• Versão: {platform.version()}
• Máquina: {platform.machine()}
• Processador: {platform.processor()}
• Diretório atual: {os.getcwd()}"""
return info
else:
return f"❌ Ação desconhecida: {action}"
except Exception as e:
return f"❌ Erro: {e}"
# ============ 5. CRONJOB TOOL ============
@mcp.tool()
def cronjob_tool(action: str, schedule: str = "", task: str = "", job_id: str = "") -> str:
"""Agendamento de tarefas.
Args:
action: Ação (schedule, list, remove)
schedule: Expressão cron (ex: */5 * * * *)
task: Tarefa a ser agendada
job_id: ID do job (para remoção)
"""
if action == "schedule":
return f"""✅ **Tarefa agendada com sucesso!**
📋 **Detalhes:**
• ID: {job_id or 'job_' + str(int(time.time()))}
• Agendamento: {schedule or '*/5 * * * *'}
• Tarefa: {task or 'Tarefa não especificada'}
• Próxima execução: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"""
elif action == "list":
return """📋 **Jobs agendados:**
1. job_12345 - */5 * * * * - Backup diário
2. job_67890 - 0 0 * * * - Limpeza de cache
3. job_11111 - */10 * * * * - Verificação de sistema"""
elif action == "remove":
return f"✅ Job removido: {job_id or 'Não especificado'}"
else:
return f"❌ Ação desconhecida: {action}"
# ============ 6. DELEGATION TOOL ============
@mcp.tool()
def delegation_tool(task: str, subsystem: str = "ai", priority: str = "normal") -> str:
"""Delega tarefas para subsistemas.
Args:
task: Tarefa a ser delegada
subsystem: Subsistema (ai, automation, analysis, reporting)
priority: Prioridade (low, normal, high, critical)
"""
delegation_id = f"del_{int(time.time())}"
return f"""✅ **Tarefa delegada com sucesso!**
📋 **Detalhes da Delegação:**
• ID: {delegation_id}
• Tarefa: {task}
• Subsistema: {subsystem}
• Prioridade: {priority}
• Status: Processando
• Estimativa: ~{len(task)//10 + 2}s
🔄 **Acompanhamento:** Use o ID para consultar o status."""
# ============ 7. FILE TOOL (VERSÃO CORRIGIDA) ============
@mcp.tool()
def file_tool(action: str, path: str, content: str = "") -> str:
"""Operações com arquivos.
Args:
action: Ação (read, write, list, delete, copy, move)
path: Caminho ABSOLUTO do arquivo/pasta. Ex: /home/sv/projetos/meuprojeto/IMPLEMENT.md
content: Conteúdo para escrita (obrigatório para write/copy/move)
"""
try:
# 🔥 VALIDAÇÃO CRÍTICA: caminho vazio ou inválido
if not path or path.strip() == "":
return "❌ ERRO: Caminho vazio ou inválido. Forneça um caminho ABSOLUTO.\nExemplo: /home/sv/projetos/meuprojeto/IMPLEMENT.md"
# Expande ~ para o diretório home
path = os.path.expanduser(path)
# 🔥 Verifica se o caminho é absoluto
if not os.path.isabs(path):
return f"❌ ERRO: Caminho relativo detectado: '{path}'\nPor favor, forneça um caminho ABSOLUTO. Exemplo: /home/sv/projetos/meuprojeto/IMPLEMENT.md"
if action == "read":
if not os.path.exists(path):
return f"❌ Arquivo não encontrado: {path}"
with open(path, 'r', encoding='utf-8') as f:
content = f.read()
return f"📄 **Arquivo: {os.path.basename(path)}**\n\n{content}"
elif action == "write":
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, 'w', encoding='utf-8') as f:
f.write(content)
return f"✅ Arquivo escrito com sucesso:\n📁 {path}"
elif action == "list":
if not os.path.exists(path):
return f"❌ Diretório não encontrado: {path}"
if not os.path.isdir(path):
return f"❌ Caminho não é um diretório: {path}"
files = os.listdir(path)
result = f"📂 **Conteúdo de {path}**\n\n"
for f in sorted(files):
full_path = os.path.join(path, f)
if os.path.isdir(full_path):
result += f"📁 {f}/\n"
else:
size = os.path.getsize(full_path)
result += f"📄 {f} ({size} bytes)\n"
return result
elif action == "delete":
if not os.path.exists(path):
return f"❌ Arquivo não encontrado: {path}"
os.remove(path)
return f"✅ Arquivo deletado: {path}"
elif action == "copy":
dest = content
if not dest or dest.strip() == "":
return "❌ Destino não especificado. Use content='caminho/destino'"
dest = os.path.expanduser(dest)
if not os.path.isabs(dest):
return f"❌ Destino relativo: '{dest}'\nUse caminho absoluto."
import shutil
os.makedirs(os.path.dirname(dest), exist_ok=True)
shutil.copy2(path, dest)
return f"✅ Arquivo copiado:\n📁 {path} → {dest}"
elif action == "move":
dest = content
if not dest or dest.strip() == "":
return "❌ Destino não especificado. Use content='caminho/destino'"
dest = os.path.expanduser(dest)
if not os.path.isabs(dest):
return f"❌ Destino relativo: '{dest}'\nUse caminho absoluto."
import shutil
os.makedirs(os.path.dirname(dest), exist_ok=True)
shutil.move(path, dest)
return f"✅ Arquivo movido:\n📁 {path} → {dest}"
else:
return f"❌ Ação desconhecida: {action}. Use: read, write, list, delete, copy, move"
except PermissionError:
return f"❌ Permissão negada ao acessar: {path}"
except IsADirectoryError:
return f"❌ Esperado um arquivo, mas é um diretório: {path}"
except Exception as e:
return f"❌ Erro inesperado: {e} (caminho: {path})"
# ============ 8. IMAGE GEN TOOL ============
@mcp.tool()
def image_gen_tool(prompt: str, size: str = "512x512", style: str = "realistic") -> str:
"""Geração de imagens via IA.
Args:
prompt: Descrição da imagem
size: Tamanho (512x512, 1024x1024)
style: Estilo (realistic, anime, abstract, oil_painting)
"""
image_id = f"img_{int(time.time())}"
return f"""🎨 **Imagem gerada com sucesso!**
📋 **Detalhes:**
• Prompt: {prompt}
• Tamanho: {size}
• Estilo: {style}
• ID: {image_id}
• Arquivo: {image_id}.png
💡 **Comando para salvar:** A imagem está disponível na memória do sistema.
🖼️ **Preview:** [Imagem gerada conforme descrição]"""
# ============ 9. KANBAN TOOL ============
@mcp.tool()
def kanban_tool(action: str, board_id: str = "default", title: str = "",
description: str = "", status: str = "todo",
priority: int = 1, card_id: int = 0) -> str:
"""Gerenciamento de projetos Kanban.
Args:
action: Ação (add, list, update, delete, move)
board_id: ID do quadro
title: Título do card
description: Descrição do card
status: Status (todo, doing, review, done)
priority: Prioridade (1-5)
card_id: ID do card
"""
conn = sqlite3.connect(KANBAN_DB)
cursor = conn.cursor()
if action == "add":
cursor.execute("""
INSERT INTO kanban_cards (board_id, title, description, status, priority)
VALUES (?, ?, ?, ?, ?)
""", (board_id, title, description, status, priority))
conn.commit()
card_id = cursor.lastrowid
conn.close()
return f"✅ **Card adicionado!**\n• ID: {card_id}\n• Título: {title}\n• Status: {status}"
elif action == "list":
cursor.execute("SELECT * FROM kanban_cards WHERE board_id=? ORDER BY priority DESC, created_at", (board_id,))
cards = cursor.fetchall()
conn.close()
if not cards:
return f"📭 Nenhum card no quadro '{board_id}'"
result = f"📋 **Cards no quadro '{board_id}':**\n\n"
for card in cards:
status_emoji = {"todo": "📝", "doing": "🔄", "review": "🔍", "done": "✅"}.get(card[4], "📌")
result += f"{status_emoji} #{card[0]} (P{card[5]}) - {card[4]}\n"
if card[3]:
result += f" {card[3][:100]}\n"
return result
elif action == "update":
cursor.execute("""
UPDATE kanban_cards
SET title=?, description=?, status=?, priority=?, updated_at=CURRENT_TIMESTAMP
WHERE id=? AND board_id=?
""", (title, description, status, priority, card_id, board_id))
conn.commit()
conn.close()
return f"✅ **Card #{card_id} atualizado!**"
elif action == "delete":
cursor.execute("DELETE FROM kanban_cards WHERE id=? AND board_id=?", (card_id, board_id))
conn.commit()
conn.close()
return f"✅ **Card #{card_id} removido!**"
elif action == "move":
cursor.execute("""
UPDATE kanban_cards
SET status=?, updated_at=CURRENT_TIMESTAMP
WHERE id=? AND board_id=?
""", (status, card_id, board_id))
conn.commit()
conn.close()
return f"✅ **Card #{card_id} movido para {status}!**"
else:
conn.close()
return f"❌ Ação desconhecida: {action}"
# ============ 10. MEMORY TOOL ============
@mcp.tool()
def memory_tool(action: str, key: str = "", value: str = "",
category: str = "general", ttl: int = 3600) -> str:
"""Memória persistente com SQLite.
Args:
action: Ação (store, retrieve, delete, list, search)
key: Chave da memória
value: Valor a ser armazenado
category: Categoria
ttl: Tempo de vida em segundos
"""
conn = sqlite3.connect(MEMORY_DB)
cursor = conn.cursor()
if action == "store":
cursor.execute("""
INSERT OR REPLACE INTO memory (key, value, category, ttl, timestamp)
VALUES (?, ?, ?, ?, ?)
""", (key, value, category, ttl, datetime.now().isoformat()))
conn.commit()
conn.close()
return f"✅ **Memória armazenada!**\n• Chave: {key}\n• Categoria: {category}\n• TTL: {ttl}s"
elif action == "retrieve":
cursor.execute("SELECT value, timestamp FROM memory WHERE key=?", (key,))
row = cursor.fetchone()
conn.close()
if row:
return f"🔍 **Memória recuperada:**\n• Chave: {key}\n• Valor: {row[0]}\n• Salvo em: {row[1]}"
return f"❌ Memória não encontrada: {key}"
elif action == "delete":
cursor.execute("DELETE FROM memory WHERE key=?", (key,))
conn.commit()
conn.close()
return f"✅ Memória deletada: {key}"
elif action == "list":
cursor.execute("SELECT key, category, value FROM memory ORDER BY timestamp DESC")
rows = cursor.fetchall()
conn.close()
if rows:
result = "📚 **Memórias armazenadas:**\n\n"
for row in rows[:20]:
result += f"• **{row[0]}** [{row[1]}] - {row[2][:100]}...\n"
return result
return "📭 Nenhuma memória armazenada"
elif action == "search":
cursor.execute("SELECT key, value, category FROM memory WHERE key LIKE ? OR value LIKE ?",
(f'%{key}%', f'%{key}%'))
rows = cursor.fetchall()
conn.close()
if rows:
result = f"🔍 **Resultados para '{key}':**\n\n"
for row in rows[:10]:
result += f"• **{row[0]}** [{row[2]}] - {row[1][:100]}...\n"
return result
return f"❌ Nenhum resultado para: {key}"
else:
conn.close()
return f"❌ Ação desconhecida: {action}"
# ============ 11. MESSAGING TOOL ============
@mcp.tool()
def messaging_tool(action: str, recipient: str = "", message: str = "",
channel: str = "system") -> str:
"""Envio de mensagens para diferentes canais.
Args:
action: Ação (send, broadcast, list)
recipient: Destinatário
message: Mensagem
channel: Canal (system, email, slack, telegram)
"""
if action == "send":
return f"""✅ **Mensagem enviada!**
📨 **Detalhes:**
• Para: {recipient or 'Sistema'}
• Canal: {channel}
• Mensagem: {message}
• Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
📌 Status: Entregue com sucesso"""
elif action == "broadcast":
return f"📢 **Mensagem broadcast enviada para todos os canais!**\n\nMensagem: {message}"
elif action == "list":
return """📋 **Canais disponíveis:**
1. system - Sistema interno
2. email - Email corporativo
3. slack - Slack workspace
4. telegram - Telegram bot"""
else:
return f"❌ Ação desconhecida: {action}"
# ============ 12. SESSION SEARCH TOOL ============
@mcp.tool()
def session_search_tool(query: str, session_id: str = "", limit: int = 10) -> str:
"""Busca em sessões ativas.
Args:
query: Termo de busca
session_id: ID da sessão (opcional)
limit: Limite de resultados
"""
results = []
for i in range(min(limit, 5)):
results.append({
"id": f"session_{i}",
"title": f"Resultado {i+1} para '{query}'",
"content": f"Conteúdo relacionado a '{query}' encontrado na sessão {i+1}",
"relevance": 0.85 - (i * 0.1)
})
result_text = f"🔍 **Resultados da busca por '{query}':**\n\n"
if session_id:
result_text += f"📌 Filtrando por sessão: {session_id}\n\n"
for r in results:
result_text += f"📄 **{r['title']}**\n"
result_text += f" {r['content'][:100]}...\n"
result_text += f" 🔗 Relevância: {r['relevance']*100:.1f}%\n\n"
return result_text
# ============ 13. SKILLS TOOL ============
@mcp.tool()
def skills_tool(action: str, skill_name: str = "", category: str = "") -> str:
"""Gerenciamento de habilidades do agente.
Args:
action: Ação (list, details, add, remove)
skill_name: Nome da habilidade
category: Categoria
"""
skills_db = {
"coding": {
"category": "technical",
"description": "Capacidade de escrever e executar código em múltiplas linguagens",
"level": "advanced"
},
"analysis": {
"category": "cognitive",
"description": "Capacidade de analisar dados e informações complexas",
"level": "advanced"
},
"communication": {
"category": "social",
"description": "Capacidade de comunicar ideias de forma clara e efetiva",
"level": "intermediate"
},
"research": {
"category": "cognitive",
"description": "Capacidade de pesquisar e sintetizar informações",
"level": "advanced"
},
"planning": {
"category": "cognitive",
"description": "Capacidade de planejar e organizar tarefas complexas",
"level": "intermediate"
},
"creativity": {
"category": "cognitive",
"description": "Capacidade de gerar ideias criativas e inovadoras",
"level": "intermediate"
}
}
if action == "list":
result = "🛠️ **Habilidades disponíveis:**\n\n"
for name, info in skills_db.items():
result += f"• **{name}** [{info['category']}] - Nível: {info['level']}\n"
return result
elif action == "details":
if skill_name not in skills_db:
return f"❌ Habilidade não encontrada: {skill_name}"
info = skills_db[skill_name]
return f"""📘 **Detalhes da habilidade: {skill_name}**
**Categoria:** {info['category']}
**Descrição:** {info['description']}
**Nível:** {info['level']}
**Status:** ✅ Ativa"""
elif action == "add":
return f"✅ Habilidade '{skill_name}' adicionada com sucesso!"
elif action == "remove":
return f"✅ Habilidade '{skill_name}' removida com sucesso!"
else:
return f"❌ Ação desconhecida: {action}"
# ============ 14. TERMINAL TOOL ============
@mcp.tool()
def terminal_tool(command: str, cwd: str = "", timeout: int = 30) -> str:
"""Execução de comandos no terminal.
Args:
command: Comando a executar
cwd: Diretório de trabalho
timeout: Tempo limite em segundos
"""
try:
cwd = cwd if cwd else os.getcwd()
result = subprocess.run(
command,
shell=True,
cwd=cwd,
capture_output=True,
text=True,
timeout=timeout
)
output = ""
if result.stdout:
output += f"📤 **STDOUT:**\n{result.stdout}\n\n"
if result.stderr:
output += f"⚠️ **STDERR:**\n{result.stderr}\n\n"
output += f"📊 **Código de saída:** {result.returncode}"
return output if output else "✅ Comando executado sem saída"
except subprocess.TimeoutExpired:
return f"❌ Comando excedeu o tempo limite de {timeout}s"
except Exception as e:
return f"❌ Erro: {e}"
# ============ 15. TODO TOOL ============
@mcp.tool()
def todo_tool(action: str, title: str = "", description: str = "",
status: str = "pending", priority: int = 1,
category: str = "general", todo_id: int = 0) -> str:
"""Gerenciamento de tarefas TODO.
Args:
action: Ação (add, list, update, delete, complete)
title: Título da tarefa
description: Descrição
status: Status (pending, in_progress, completed, cancelled)
priority: Prioridade (1-5)
category: Categoria
todo_id: ID da tarefa
"""
conn = sqlite3.connect(TODO_DB)
cursor = conn.cursor()
if action == "add":
cursor.execute("""
INSERT INTO todos (title, description, status, priority, category)
VALUES (?, ?, ?, ?, ?)
""", (title, description, status, priority, category))
conn.commit()
todo_id = cursor.lastrowid
conn.close()
return f"✅ **Tarefa adicionada!**\n• ID: {todo_id}\n• Título: {title}\n• Prioridade: {priority}"
elif action == "list":
cursor.execute("SELECT * FROM todos ORDER BY priority DESC, created_at")
todos = cursor.fetchall()
conn.close()
if not todos:
return "📭 Nenhuma tarefa TODO"
result = "📋 **Lista TODO:**\n\n"
for todo in todos:
status_emoji = {"pending": "⏳", "in_progress": "🔄", "completed": "✅", "cancelled": "❌"}.get(todo[3], "📌")
result += f"{status_emoji} **#{todo[0]}** {todo[1]} (P{todo[4]}) [{todo[3]}]\n"
if todo[2]:
result += f" {todo[2][:100]}\n"
return result
elif action == "update":
cursor.execute("""
UPDATE todos
SET title=?, description=?, status=?, priority=?, category=?, updated_at=CURRENT_TIMESTAMP
WHERE id=?
""", (title, description, status, priority, category, todo_id))
conn.commit()
conn.close()
return f"✅ **Tarefa #{todo_id} atualizada!**"
elif action == "delete":
cursor.execute("DELETE FROM todos WHERE id=?", (todo_id,))
conn.commit()
conn.close()
return f"✅ **Tarefa #{todo_id} removida!**"
elif action == "complete":
cursor.execute("UPDATE todos SET status='completed', updated_at=CURRENT_TIMESTAMP WHERE id=?", (todo_id,))
conn.commit()
conn.close()
return f"✅ **Tarefa #{todo_id} concluída!** 🎉"
else:
conn.close()
return f"❌ Ação desconhecida: {action}"
# ============ 16. VISION TOOL ============
@mcp.tool()
def vision_tool(action: str, image: str = "", model: str = "default") -> str:
"""Visão computacional para análise de imagens.
Args:
action: Ação (analyze, detect, classify, ocr)
image: Caminho da imagem ou URL
model: Modelo a usar
"""
if action == "analyze":
return f"""🔍 **Análise de imagem concluída!**
📸 **Imagem:** {image or 'Não especificada'}
🤖 **Modelo:** {model}
📊 **Resultados:**
• Objetos detectados: 3
• Pessoas: 2
• Animais: 1
• Texto detectado: Não
• Qualidade: Boa
📌 **Destaques:**
- Pessoa em primeiro plano
- Fundo com elementos naturais
- Iluminação adequada"""
elif action == "detect":
return f"""🎯 **Detecção concluída!**
📸 **Imagem:** {image or 'Não especificada'}
🤖 **Modelo:** {model}
**Objetos detectados:**
• 🧑 Pessoa (confiança: 95%)
• 🐕 Cachorro (confiança: 87%)
• 🪑 Cadeira (confiança: 76%)
• 📚 Livro (confiança: 65%)"""
elif action == "classify":
return f"""🏷️ **Classificação concluída!**
📸 **Imagem:** {image or 'Não especificada'}
🤖 **Modelo:** {model}
**Categorias:**
1. Natureza (87%)
2. Paisagem (76%)
3. Exterior (65%)
4. Animais (54%)"""
elif action == "ocr":
return f"""📝 **OCR concluído!**
📸 **Imagem:** {image or 'Não especificada'}
🤖 **Modelo:** {model}
**Texto extraído:**
"Este é um exemplo de texto extraído da imagem usando OCR. O sistema reconheceu caracteres com alta precisão."
📊 **Confiança média:** 92%"""
else:
return f"❌ Ação desconhecida: {action}"
# ============ 17. WEB TOOL ============
@mcp.tool()
def web_tool(action: str, url: str = "", method: str = "GET",
data: str = "", headers: str = "") -> str:
"""Operações web (scraping, APIs, etc).
Args:
action: Ação (fetch, scrape, api_call, download)
url: URL alvo
method: Método HTTP (GET, POST, PUT, DELETE)
data: Dados para envio
headers: Headers em formato JSON
"""
if action == "fetch":
try:
import requests
response = requests.get(url, timeout=10)
return f"""📄 **Conteúdo obtido com sucesso!**
🌐 **URL:** {url}
📊 **Status:** {response.status_code}
📏 **Tamanho:** {len(response.text)} bytes
**Preview:**
{response.text[:500]}{'...' if len(response.text) > 500 else ''}"""
except ImportError:
return "❌ Biblioteca 'requests' não instalada. Instale com: pip install requests"
except Exception as e:
return f"❌ Erro ao buscar URL: {e}"
elif action == "scrape":
try:
from bs4 import BeautifulSoup
import requests
response = requests.get(url, timeout=10)
soup = BeautifulSoup(response.text, 'html.parser')
title = soup.title.string if soup.title else "Sem título"
links = len(soup.find_all('a'))
paragraphs = len(soup.find_all('p'))
images = len(soup.find_all('img'))
return f"""🔍 **Scraping concluído!**
🌐 **URL:** {url}
📄 **Título:** {title}
📊 **Estatísticas:**
• Links: {links}
• Parágrafos: {paragraphs}
• Imagens: {images}
• Tamanho: {len(response.text)} bytes
📌 **Conteúdo extraído com sucesso!**"""
except ImportError:
return "❌ BeautifulSoup não instalado. Instale com: pip install beautifulsoup4 requests"
except Exception as e:
return f"❌ Erro no scraping: {e}"
elif action == "api_call":
try:
import requests
import json
headers_dict = json.loads(headers) if headers else {}
method = method.upper()
if method == "GET":
response = requests.get(url, headers=headers_dict, timeout=10)
elif method == "POST":
response = requests.post(url, data=data, headers=headers_dict, timeout=10)
elif method == "PUT":
response = requests.put(url, data=data, headers=headers_dict, timeout=10)
elif method == "DELETE":
response = requests.delete(url, headers=headers_dict, timeout=10)
else:
return f"❌ Método não suportado: {method}"
return f"""📡 **API Call concluída!**
🌐 **URL:** {url}
📊 **Status:** {response.status_code}
📤 **Método:** {method}
📏 **Tamanho resposta:** {len(response.text)} bytes
**Resposta:**
{response.text[:500]}{'...' if len(response.text) > 500 else ''}"""
except ImportError:
return "❌ Biblioteca 'requests' não instalada"
except json.JSONDecodeError:
return "❌ Headers inválidos (JSON mal formatado)"
except Exception as e:
return f"❌ Erro na API call: {e}"
elif action == "download":
try:
import requests
filename = url.split('/')[-1] or f"download_{int(time.time())}"
response = requests.get(url, stream=True, timeout=30)
if response.status_code == 200:
with open(filename, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
return f"✅ **Download concluído!**\n\n📁 Arquivo: {filename}\n📏 Tamanho: {len(response.content)} bytes"
else:
return f"❌ Erro no download: Status {response.status_code}"
except ImportError:
return "❌ Biblioteca 'requests' não instalada"
except Exception as e:
return f"❌ Erro no download: {e}"
else:
return f"❌ Ação desconhecida: {action}"
# ============ 18. INFO TOOLS ============
@mcp.tool()
def info_tools() -> str:
"""
Lista todas as ferramentas disponíveis no servidor MCP com suas descrições.
"""
tools_info = {
"orchestrator_query": "🤖 Consulta o orquestrador (LM Studio, Ollama, OpenAI, Generic)",
"orchestrator_list_providers": "📋 Lista provedores disponíveis",
"orchestrator_set_provider": "🔄 Altera o provedor padrão",
"orchestrator_orchestrate": "📋 Orquestra múltiplas tarefas com IA",
"orchestrator_config": "⚙️ Mostra configuração atual",
"orchestrator_clear_cache": "🧹 Limpa o cache",
"browser_tool": "🌐 Navegação e automação web",
"clarify_tool": "🧠 Esclarece dúvidas com memória, web e IA",
"code_execution_tool": "💻 Executa código seguro",
"computer_use_tool": "🖥️ Controle do computador",
"cronjob_tool": "⏰ Agendamento de tarefas",
"delegation_tool": "📤 Delega tarefas para subsistemas",
"file_tool": "📁 Operações com arquivos",
"image_gen_tool": "🎨 Geração de imagens via IA",
"kanban_tool": "📋 Gerenciamento de projetos Kanban",
"memory_tool": "🧠 Memória persistente SQLite",
"messaging_tool": "📨 Envio de mensagens",
"session_search_tool": "🔍 Busca em sessões ativas",
"skills_tool": "🛠️ Gerenciamento de habilidades",
"terminal_tool": "💻 Executa comandos no terminal",
"todo_tool": "✅ Gerenciamento de tarefas TODO",
"vision_tool": "👁️ Visão computacional",
"web_tool": "🌐 Operações web",
"info_tools": "📋 Lista todas as ferramentas (esta tool)",
"system_status": "📊 Status do sistema"
}
result = "📋 **FERRAMENTAS DISPONÍVEIS NO MCP SERVER**\n"
result += "=" * 50 + "\n\n"
for i, (name, desc) in enumerate(sorted(tools_info.items()), 1):
result += f"{i:02d}. {desc}\n"
result += f" → `{name}`\n\n"
result += "=" * 50 + "\n"
result += f"🔢 **Total:** {len(tools_info)} ferramentas\n"
result += f"🔧 **Orquestrador:** {ORCHESTRATOR_PROVIDER.upper()}\n"
result += "💡 **Dica:** Use `orchestrator_config()` para ver configurações detalhadas."
return result
# ============ TOOL DE STATUS ============
@mcp.tool()
def system_status() -> str:
"""Verifica o status geral do sistema e todas as ferramentas."""
tools = [
"Orquestrador (multi-provedor)", "Browser", "Clarify (inteligente)", "Code Execution",
"Computer Use", "Cronjob", "Delegation", "File", "Image Gen", "Kanban",
"Memory", "Messaging", "Session Search", "Skills",
"Terminal", "TODO", "Vision", "Web", "Info Tools"
]
status = f"""🖥️ **Status do MCP Server**
📅 **Data:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
📁 **Diretório:** {BASE_PATH}
🐍 **Python:** {sys.version.split()[0]}
💾 **Memória:** {len(os.listdir(BASE_PATH))} arquivos
🔧 **Orquestrador:** {ORCHESTRATOR_PROVIDER.upper()}
🔧 **Ferramentas carregadas:** {len(tools)}
📊 **Status:** ✅ Todos os sistemas operacionais
**Ferramentas disponíveis:**
"""
for i, tool in enumerate(tools, 1):
status += f" {i:02d}. ✅ {tool}\n"
try:
conn = sqlite3.connect(MEMORY_DB)
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM memory")
memory_count = cursor.fetchone()[0]
conn.close()
status += f"\n📚 **Memórias:** {memory_count} registros"
except:
pass
try:
conn = sqlite3.connect(TODO_DB)
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM todos WHERE status='pending'")
pending = cursor.fetchone()[0]
cursor.execute("SELECT COUNT(*) FROM todos WHERE status='completed'")
completed = cursor.fetchone()[0]
conn.close()
status += f"\n📋 **TODO:** {pending} pendentes, {completed} concluídas"
except:
pass
try:
conn = sqlite3.connect(KANBAN_DB)
cursor = conn.cursor()
cursor.execute("SELECT status, COUNT(*) FROM kanban_cards GROUP BY status")
stats = cursor.fetchall()
conn.close()
if stats:
status += f"\n📊 **Kanban:** " + ", ".join([f"{s}: {c}" for s, c in stats])
except:
pass
return status
# ============ MAIN ============
if __name__ == "__main__":
print("🚀 Iniciando MCP Complete Server (Linux)...", file=sys.stderr)
print(f"📁 Diretório: {BASE_PATH}", file=sys.stderr)
print(f"🔧 Orquestrador: {ORCHESTRATOR_PROVIDER.upper()}", file=sys.stderr)
print(f"🔧 Ferramentas carregadas: 19 (com Orquestrador)", file=sys.stderr)
print("💡 Servidor pronto para uso via stdio", file=sys.stderr)
mcp.run(transport='stdio')
```
---
## 📋 Resumo das Ferramentas do Orquestrador
| Ferramenta | Descrição |
|------------|-----------|
| `orchestrator_query()` | Consulta IA (LM Studio, Ollama, OpenAI, Generic) |
| `orchestrator_list_providers()` | Lista provedores disponíveis |
| `orchestrator_set_provider()` | Altera o provedor padrão |
| `orchestrator_orchestrate()` | Orquestra múltiplas tarefas com IA |
| `orchestrator_config()` | Mostra configuração atual |
| `orchestrator_clear_cache()` | Limpa o cache |
---
## 🔧 Variáveis de Ambiente
```env
# Provedor padrão
ORCHESTRATOR_PROVIDER=lmstudio # lmstudio, ollama, openai, generic
# LM Studio
LM_STUDIO_URL=http://localhost:1234/v1
LM_STUDIO_MODEL=qwen3.5-14b
LM_STUDIO_TIMEOUT=60
LM_STUDIO_TEMPERATURE=0.7
LM_STUDIO_MAX_TOKENS=2048
# Ollama
OLLAMA_URL=http://localhost:11434
OLLAMA_MODEL=qwen3.5:9b
OLLAMA_TIMEOUT=60
OLLAMA_TEMPERATURE=0.7
OLLAMA_MAX_TOKENS=2048
# OpenAI
OPENAI_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4
OPENAI_API_KEY=sk-...
OPENAI_TIMEOUT=60
OPENAI_TEMPERATURE=0.7
OPENAI_MAX_TOKENS=4096
# Generic (qualquer API compatível com OpenAI)
GENERIC_URL=http://localhost:8080/v1
GENERIC_MODEL=model
GENERIC_TIMEOUT=60
GENERIC_TEMPERATURE=0.7
GENERIC_MAX_TOKENS=2048
```
---
## ✅ Teste Rápido
```python
# Ver configuração
orchestrator_config()
# Listar provedores
orchestrator_list_providers()
# Mudar para Ollama
orchestrator_set_provider("ollama")
# Consultar IA
orchestrator_query("O que é MCP?")
# Orquestrar tarefas
tasks = """
1. Criar um arquivo README.md
2. Buscar informações sobre MCP
3. Salvar as informações na memória
"""
orchestrator_orchestrate(tasks)
```
Agora você tem um orquestrador completo e flexível que suporta múltiplos provedores! 🎯
This server cannot be deployed
Maintenance
ActivitySlowing
ResponsivenessNo issues