# unknown

> Segue um **repo skeleton completo (copiar/colar)** no padrão “DevOps sênior”, já alinhado com tudo que a gente combinou: LangGraph + **memória 3 níveis** + **batching** + **Router MCP Tools** + **RAG Qdrant** + **LLM Router (LiteLLM)** e pronto para rodar como **Application no Coolify** consumindo seus **Services**. [github](https://github.com/wassim249/fastapi-langgraph-agent-production-ready-template) ## Estrutura do repositório `zappro-core` ```txt zappro-core/ ├─ README.md ├─ SKILL.md ├─ ...

- Author: Will.zappro
- Repository: zapprosite/zappro-mvp
- Version: 20260124051708
- Stars: 0
- Forks: 0
- Last Updated: 2026-02-07
- Source: https://github.com/zapprosite/zappro-mvp
- Web: https://mule.run/skillshub/@@zapprosite/zappro-mvp~unknown:20260124051708

---

Segue o **Patch 1 + Patch 2 + Patch 3** em modo “copiar/colar” (arquivos completos), mantendo seu skeleton e elevando para “nível wassim” no que importa: **observabilidade (logs+métricas)**, **rate limit**, e **ToolNode/tools_condition** para o Router MCP. [github](https://github.com/wassim249/fastapi-langgraph-agent-production-ready-template)
> Convenções usadas abaixo  
> - Seus módulos seguem `app/...`  
> - O FastAPI fica em `app/main.py` (como você já tem).  
> - Redis já existe (memória curta), então reaproveitamos para rate limit.  
> - MCP tools aqui são “stubs” (HTTP) para você ligar depois no seu MetaMCP/serviços.

***

## PATCH 1 — Observabilidade (logs + métricas)
### 1) `app/core/logging.py`
```python
import json
import logging
import os
import time
import uuid
from contextvars import ContextVar

request_id_var: ContextVar[str] = ContextVar("request_id", default="")
user_id_var: ContextVar[str] = ContextVar("user_id", default="")
thread_id_var: ContextVar[str] = ContextVar("thread_id", default="")

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": int(time.time() * 1000),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
            "request_id": request_id_var.get() or None,
            "user_id": user_id_var.get() or None,
            "thread_id": thread_id_var.get() or None,
        }
        if record.exc_info:
            payload["exc_info"] = self.formatException(record.exc_info)
        return json.dumps(payload, ensure_ascii=False)

def setup_logging():
    level = os.getenv("LOG_LEVEL", "INFO").upper()
    root = logging.getLogger()
    root.setLevel(level)

    handler = logging.StreamHandler()
    handler.setFormatter(JsonFormatter())
    root.handlers = [handler]

def new_request_id() -> str:
    return uuid.uuid4().hex
```

### 2) `app/core/metrics.py`
```python
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST

REQ_COUNT = Counter(
    "zappro_requests_total",
    "Total requests",
    ["path", "method", "status"]
)

REQ_LATENCY = Histogram(
    "zappro_request_latency_seconds",
    "Request latency in seconds",
    ["path", "method"]
)

LLM_COUNT = Counter(
    "zappro_llm_calls_total",
    "Total LLM calls",
    ["provider", "model"]
)

TOOLS_COUNT = Counter(
    "zappro_tool_calls_total",
    "Total tool calls",
    ["tool_name"]
)

def metrics_response():
    data = generate_latest()
    return data, CONTENT_TYPE_LATEST
```

### 3) `requirements.txt` (adicione)
```txt
prometheus-client==0.20.0
```

### 4) Middleware de observabilidade em `app/main.py` (trecho)
Cole isso no seu `app/main.py`:

```python
import time
from fastapi import Request, Response
from app.core.logging import setup_logging, new_request_id, request_id_var, user_id_var, thread_id_var
from app.core.metrics import REQ_COUNT, REQ_LATENCY, metrics_response

setup_logging()

@app.middleware("http")
async def observability_middleware(request: Request, call_next):
    rid = new_request_id()
    request_id_var.set(rid)

    start = time.time()
    try:
        resp: Response = await call_next(request)
        status = str(resp.status_code)
        return resp
    finally:
        elapsed = time.time() - start
        REQ_LATENCY.labels(path=request.url.path, method=request.method).observe(elapsed)
        # status pode não existir em exceções: usa "500"
        try:
            REQ_COUNT.labels(path=request.url.path, method=request.method, status=str(resp.status_code)).inc()
        except Exception:
            REQ_COUNT.labels(path=request.url.path, method=request.method, status="500").inc()

@app.get("/metrics")
async def metrics():
    data, ctype = metrics_response()
    return Response(content=data, media_type=ctype)
```

***

## PATCH 2 — Rate limit (Redis) no `/api/chat`
### 1) `app/core/rate_limit.py`
```python
import time
import redis
from dataclasses import dataclass

@dataclass
class RateLimitResult:
    allowed: bool
    remaining: int
    reset_seconds: int

class RedisRateLimiter:
    """
    Token-bucket simplificado por janela fixa (rolling-ish):
    - Key = rl:{scope}:{identifier}:{window_start}
    - Permite N requests por janela (segundos)
    """
    def __init__(self, redis_url: str):
        self.r = redis.from_url(redis_url)

    def hit(self, scope: str, identifier: str, limit: int, window_seconds: int) -> RateLimitResult:
        now = int(time.time())
        window_start = now - (now % window_seconds)
        key = f"rl:{scope}:{identifier}:{window_start}"

        count = self.r.incr(key)
        if count == 1:
            self.r.expire(key, window_seconds)

        allowed = count <= limit
        remaining = max(0, limit - count)
        reset_seconds = window_seconds - (now % window_seconds)
        return RateLimitResult(allowed=allowed, remaining=remaining, reset_seconds=reset_seconds)
```

### 2) Aplicar no endpoint `/api/chat` (ou middleware)
No seu `app/main.py`, faça assim (bem simples e controlável):

```python
from fastapi import HTTPException
from app.core.config import settings
from app.core.rate_limit import RedisRateLimiter

rate_limiter = RedisRateLimiter(settings.REDIS_URL)

# Exemplo: 60 req/min por user_id, e fallback por IP se user_id não vier
CHAT_LIMIT_PER_MIN = int(getattr(settings, "CHAT_LIMIT_PER_MIN", 60))
CHAT_WINDOW_SECONDS = 60
```

Dentro do endpoint `/api/chat`, antes de chamar o graph:

```python
identifier = payload.user_id or "anon"
rl = rate_limiter.hit("chat", identifier, limit=CHAT_LIMIT_PER_MIN, window_seconds=CHAT_WINDOW_SECONDS)
if not rl.allowed:
    raise HTTPException(
        status_code=429,
        detail=f"Rate limit excedido. Tente novamente em {rl.reset_seconds}s."
    )
```

E no `.env.example`, você pode adicionar:

```env
CHAT_LIMIT_PER_MIN=60
```

***

## PATCH 3 — Router MCP Tools com ToolNode + tools_condition
A ideia aqui é:  
1) você cria tools (mesmo que sejam HTTP stubs agora),  
2) usa `.bind_tools(tools)` no modelo,  
3) usa **ToolNode(tools)** e uma **condição** para decidir se chama tool ou não. [reference.langchain](https://reference.langchain.com/python/langgraph/agents/)

### 1) `app/mcp/tools.py`
```python
import httpx
from typing import Any, Dict
from langchain_core.tools import tool
from app.core.config import settings

async def _call_mcp(path: str, payload: Dict[str, Any]) -> Dict[str, Any]:
    url = f"{settings.MCP_REGISTRY_URL.rstrip('/')}/{path.lstrip('/')}"
    timeout = settings.MCP_TIMEOUT_SECONDS
    async with httpx.AsyncClient(timeout=timeout) as client:
        r = await client.post(url, json=payload)
        r.raise_for_status()
        return r.json()

@tool
async def ucp_catalog(query: str) -> str:
    """Busca produto/peça no catálogo (MCP)."""
    data = await _call_mcp("/tools/ucp_catalog", {"query": query})
    return data.get("result", str(data))

@tool
async def ucp_checkout(item_id: str, qty: int = 1) -> str:
    """Consulta estoque/preço e simula checkout (MCP)."""
    data = await _call_mcp("/tools/ucp_checkout", {"item_id": item_id, "qty": qty})
    return data.get("result", str(data))

@tool
async def agenda_consulta(date_hint: str = "") -> str:
    """Consulta disponibilidade de agenda (MCP)."""
    data = await _call_mcp("/tools/agenda_consulta", {"date_hint": date_hint})
    return data.get("result", str(data))

@tool
async def agenda_cria(title: str, date_iso: str) -> str:
    """Cria agendamento (MCP)."""
    data = await _call_mcp("/tools/agenda_cria", {"title": title, "date_iso": date_iso})
    return data.get("result", str(data))
```

### 2) `app/mcp/bind.py`
```python
from app.mcp.tools import ucp_catalog, ucp_checkout, agenda_consulta, agenda_cria

def get_tools():
    return [ucp_catalog, ucp_checkout, agenda_consulta, agenda_cria]
```

### 3) `app/llm/litellm_client.py` (cliente LLM via LiteLLM proxy)
```python
import httpx
from typing import List
from langchain_core.messages import BaseMessage

from app.core.config import settings
from app.core.metrics import LLM_COUNT

def _to_openai_messages(messages: List[BaseMessage]):
    out = []
    for m in messages:
        role = getattr(m, "type", None) or m.__class__.__name__.lower()
        # LangChain messages têm .type: "human"/"ai"/"system"
        if role == "human":
            role = "user"
        elif role == "ai":
            role = "assistant"
        elif role == "system":
            role = "system"
        out.append({"role": role, "content": m.content})
    return out

async def litellm_chat(messages: List[BaseMessage], user_id: str, model: str = "auto", max_tokens: int = 800):
    payload = {
        "model": model,
        "messages": _to_openai_messages(messages),
        "max_tokens": max_tokens,
        "user": user_id,
    }
    async with httpx.AsyncClient(timeout=90) as client:
        r = await client.post(f"{settings.LITELLM_URL.rstrip('/')}/v1/chat/completions", json=payload)
        r.raise_for_status()
        data = r.json()

    used_model = data.get("model", model)
    LLM_COUNT.labels(provider="litellm", model=used_model).inc()
    return data["choices"][0]["message"]["content"]
```

### 4) `app/agent/graph.py` (com ToolNode e roteamento)
Aqui vai uma versão “mínima porém correta” do loop tool-calling:

```python
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI

from app.agent.state import AgentState
from app.core.config import settings
from app.mcp.bind import get_tools
from app.rag.retriever import rag_retrieve
from app.memory.short_redis import short_load, short_save
from app.memory.mid_postgres import mid_load, mid_save
from app.memory.long_mongo import long_load

# Observação: aqui eu uso ChatOpenAI só como "engine" de tool-calling.
# Você pode trocar por um wrapper que chame LiteLLM, mas o mais simples:
# - Tool calling com ChatOpenAI
# - Resposta final com LiteLLM
# (depois a gente unifica)
llm_tool = ChatOpenAI(model="gpt-4o-mini", temperature=0)  # placeholder
tools = get_tools()
llm_tool = llm_tool.bind_tools(tools)

tool_node = ToolNode(tools)

async def node_load_memory(state: AgentState):
    user_profile = await long_load(state["user_id"])
    session = await mid_load(state["user_id"], state["thread_id"])
    short_msgs = await short_load(state["user_id"], state["thread_id"])
    return {
        "user_profile": user_profile,
        "session_summary": session.get("summary", ""),
        "session_entities": session.get("entities", {}),
        "messages": short_msgs,
    }

async def node_rag(state: AgentState):
    q = state["messages"][-1].content
    ctx = await rag_retrieve(q, top_k=5)
    return {"rag_context": ctx}

async def node_router_llm(state: AgentState):
    system = f"""Você é um assistente HVAC-R do ZapPRO.
Quando o usuário pedir ação (estoque, agenda, catálogo, orçamento), use tools.
Se for só dúvida técnica, responda direto.

Perfil: {state.get('user_profile', {})}
Sessão: {state.get('session_summary', '')}

RAG:
{state.get('rag_context','')}
"""
    msgs = [SystemMessage(content=system), *state["messages"][-10:]]
    # Esse LLM vai decidir se precisa tool (gera tool_calls) ou responder direto
    resp = await llm_tool.ainvoke(msgs)
    return {"messages": [resp]}

async def node_save_memory(state: AgentState):
    await short_save(state["user_id"], state["thread_id"], state["messages"])
    await mid_save(state["user_id"], state["thread_id"], state)
    return {}

def build_graph():
    g = StateGraph(AgentState)
    g.add_node("load_memory", node_load_memory)
    g.add_node("rag", node_rag)
    g.add_node("router_llm", node_router_llm)
    g.add_node("tools", tool_node)
    g.add_node("save_memory", node_save_memory)

    g.add_edge(START, "load_memory")
    g.add_edge("load_memory", "rag")
    g.add_edge("rag", "router_llm")

    # Decisão: se o LLM retornou tool_calls -> vai para ToolNode, senão vai pro save_memory
    g.add_conditional_edges("router_llm", tools_condition, {
        "tools": "tools",
        END: "save_memory",
    })

    # Depois de executar tools, volta para o router_llm (loop) para continuar
    g.add_edge("tools", "router_llm")

    g.add_edge("save_memory", END)
    return g.compile()
```

**Nota importante:** o `tools_condition` e `ToolNode` são o padrão “oficial” para tool-calling em LangGraph. [reference.langchain](https://reference.langchain.com/python/langgraph/agents/)

> Depois a gente troca o `ChatOpenAI` do tool-calling para também bater no seu LiteLLM (se seu provider suportar tool calling bem). Por enquanto isso te dá o “Router MCP” real funcionando.

***

## Ajuste pequeno no seu README
Remova as tags tipo `[web:96]` do README do repo e troque por links/documentação, porque isso é notação nossa, não do GitHub.

***

## Quer que eu escreva o PATCH 4 também (adapter Telegram com batching)?
Você pediu Patch 1–3, já está entregue.  
Se você disser “manda o Patch 4”, eu te entrego:
- `adapters/telegram/webhook.py` + `batching/aggregator.py` usando Redis + janela de 2.5s, no estilo “junta e responde uma vez”.