自主AI代理服务器

一个基于Model Context Protocol (MCP)的自主AI代理服务器,能够同步处理SQLite数据库任务和异步调用外部API,适用于混合Windows和WSL环境。

  • by AirtonLira
  • Python
  • 本地部署
  • 2 星标
  • 更新于
在 GitHub 查看 AirtonLira/mcp_learning

使用说明

🤖 Super MCP Agent:完整的项目文档

本文整合了整个学习过程,以使用 **模型上下文协议(MCP)**最终项目包括一个统一的“超级服务器”,能够在Windows + WSL混合环境中编排同步(SQLite数据库)和异步(外部API)任务。


🏗️ 系统架构

  • 主机(客户端): 克劳德桌面(Windows)
  • 服务器 : Python 3.10+ 在 WSL 2 上运行 (Ubuntu)
  • 协议 : MCP(Model Context Protocol)
  • 数据库 : SQLite(本地持久性)
  • 网络 : httpx Async(与Open-Meteo API集成)

📂 推荐的文件结构

mcp_learning/
├── .venv/                  # Ambiente Virtual Python
├── loja.db                 # Banco de Dados de Produtos (Gerado pelo script)
├── super_server.py         # O CÓDIGO FINAL (Unificado)
├── criar_banco.py          # Script de setup do banco
└── README.md               # Este arquivo

💻 Códigos Fonte
1. Script de Setup do Banco (criar_banco.py)
Execute este script uma vez para gerar o arquivo loja.db.

import sqlite3

def setup_database():
    conn = sqlite3.connect("loja.db")
    cursor = conn.cursor()
    
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS produtos (
            id INTEGER PRIMARY KEY,
            nome TEXT NOT NULL,
            preco REAL NOT NULL,
            estoque INTEGER NOT NULL
        )
    """)
    
    cursor.execute("SELECT count(*) FROM produtos")
    if cursor.fetchone()[0] == 0:
        dados = [
            ("Notebook Gamer", 4500.00, 10),
            ("Mouse Sem Fio", 120.50, 50),
            ("Monitor 4K", 1800.00, 15),
            ("Teclado Mecânico", 350.00, 30),
            ("Cadeira Ergonômica", 850.00, 5)
        ]
        cursor.executemany("INSERT INTO produtos (nome, preco, estoque) VALUES (?, ?, ?)", dados)
        conn.commit()
        print("Banco 'loja.db' criado com sucesso!")
    else:
        print("Banco já existe.")

    conn.close()

if __name__ == "__main__":
    setup_database()



2. O Super Servidor (super_server.py)
Este é o código principal que contém Tools (Sync/Async) e Prompts.

import sqlite3
import os
import httpx
from mcp.server.fastmcp import FastMCP

# Inicialização do Agente
mcp = FastMCP("Assistente Geral (Loja + Clima)")

# Configuração de Caminho Absoluto (CRÍTICO PARA O CLAUDE)
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DB_PATH = os.path.join(BASE_DIR, "loja.db")

# URLs da API
GEO_URL = "[https://geocoding-api.open-meteo.com/v1/search](https://geocoding-api.open-meteo.com/v1/search)"
WEATHER_URL = "[https://api.open-meteo.com/v1/forecast](https://api.open-meteo.com/v1/forecast)"

# --- BLOCO 1: FERRAMENTAS DE ESTOQUE (Banco de Dados / Sync) ---

@mcp.tool()
def listar_produtos() -> str:
    """Lista todos os produtos do estoque com preços e quantidades."""
    with sqlite3.connect(DB_PATH) as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT id, nome, preco, estoque FROM produtos")
        items = cursor.fetchall()
        
    if not items:
        return "Nenhum produto encontrado."
    
    resultado = "ID | Produto | Preço (R$) | Estoque\n"
    resultado += "-" * 40 + "\n"
    for item in items:
        resultado += f"{item[0]} | {item[1]} | {item[2]:.2f} | {item[3]}\n"
    return resultado

@mcp.tool()
def vender_produto(nome_exato: str, quantidade: int) -> str:
    """Registra uma venda e abate do estoque no banco de dados."""
    with sqlite3.connect(DB_PATH) as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT estoque FROM produtos WHERE nome = ?", (nome_exato,))
        res = cursor.fetchone()
        
        if not res:
            return f"Erro: Produto '{nome_exato}' não encontrado."
        
        estoque_atual = res[0]
        if estoque_atual < quantidade:
            return f"Estoque insuficiente. Restam apenas {estoque_atual}."
        
        novo_estoque = estoque_atual - quantidade
        cursor.execute("UPDATE produtos SET estoque = ? WHERE nome = ?", (novo_estoque, nome_exato))
        conn.commit()
        
    return f"Venda realizada! Saldo de '{nome_exato}': {novo_estoque}."

# --- BLOCO 2: FERRAMENTAS DE CLIMA (API Externa / Async) ---

@mcp.tool()
async def obter_previsao(cidade: str) -> str:
    """Consulta API externa para ver o clima atual (Async)."""
    async with httpx.AsyncClient() as client:
        try:
            # Busca Lat/Lon
            resp_geo = await client.get(GEO_URL, params={"name": cidade, "count": 1, "language": "pt"})
            resp_geo.raise_for_status()
            data_geo = resp_geo.json()
            
            if "results" not in data_geo:
                return f"Cidade '{cidade}' não encontrada."
                
            local = data_geo["results"][0]
            
            # Busca Clima
            params_clima = {
                "latitude": local["latitude"],
                "longitude": local["longitude"],
                "current": ["temperature_2m", "relative_humidity_2m"],
                "timezone": "auto"
            }
            resp_weather = await client.get(WEATHER_URL, params=params_clima)
            data_weather = resp_weather.json()
            curr = data_weather["current"]
            
            return (f"Clima em {local['name']}: {curr['temperature_2m']}°C, "
                    f"Umidade: {curr['relative_humidity_2m']}%")
            
        except Exception as e:
            return f"Erro na conexão: {str(e)}"

# --- BLOCO 3: PROMPTS ---

@mcp.prompt()
def assistente_vendas() -> str:
    """Prompt pronto para atuar como vendedor proativo."""
    return """
    Você é um assistente de vendas inteligente.
    Sua missão é:
    1. Verificar o clima da cidade do usuário.
    2. Sugerir produtos do estoque que combinem com o clima.
    """

if __name__ == "__main__":
    mcp.run()


⚙️ Instalação e Configuração
1. Dependências (WSL)
No terminal WSL (Ubuntu):


uv init
uv add "mcp[cli]" httpx
python criar_banco.py


2. Configuração do Claude Desktop (Windows)
Arquivo: %APPDATA%\Claude\claude_desktop_config.json

Substitua SEU_USUARIO pelo seu usuário do Linux.


{
  "mcpServers": {
    "super-servidor": {
      "command": "wsl.exe",
      "args": [
        "/home/SEU_USUARIO/Estudos/mcp_learning/.venv/bin/python",
        "/home/SEU_USUARIO/Estudos/mcp_learning/super_server.py"
      ]
    }
  }
}


🐛 Troubleshooting: Lições Aprendidas
Durante o desenvolvimento, resolvemos os seguintes problemas críticos:

Bug do Caminho Relativo (SQLite):

Problema: O banco loja.db não era encontrado ou era criado vazio na pasta errada.

Solução: Uso de os.path.abspath(__file__) no código Python para forçar o caminho absoluto.

Bug da Conexão WSL:

Problema: Claude Desktop não conseguia rodar comandos uv ou python do Linux.

Solução: Configuração do JSON usando wsl.exe como ponte explícita.

Bug da "Preguiça" do Modelo:

Problema: O Claude usava a previsão do tempo do Google em vez da nossa ferramenta.

Solução: Uso de prompts explícitos ("Use EXCLUSIVAMENTE a ferramenta obter_previsao").

✅ Prompt de Validação Final
Para testar se tudo está conectado, cole isto no Claude:

"Utilize EXCLUSIVAMENTE a ferramenta obter_previsao para checar a temperatura de 'Londres'. Baseado APENAS no retorno dela, se estiver frio (<15°C), use listar_produtos para checar o estoque de 'Teclado Mecânico' e então use vender_produto para vender 1 unidade."

Autor: Airton Lira Junior - https://www.linkedin.com/in/airton-de-souza-lira-junior-6b81a661/```