Compare commits

..

9 Commits

35 changed files with 3331 additions and 579 deletions

164
.env Normal file
View File

@@ -0,0 +1,164 @@
###############################################################################
# AI AGENT PLATFORM - CONFIGURAÇÃO ÚNICA
# Este arquivo é lido por Pydantic Settings no framework e no backend template.
###############################################################################
APP_NAME=ai-agent-template
APP_ENV=local
LOG_LEVEL=INFO
API_HOST=0.0.0.0
API_PORT=8000
CORS_ORIGINS=http://localhost:5173,http://127.0.0.1:5173
###############################################################################
# LLM - OCI Generative AI como provider principal
###############################################################################
# Opções: mock, oci_openai, oci_sdk, openai_compatible
LLM_PROVIDER=oci_openai
LLM_TEMPERATURE=0.2
LLM_MAX_TOKENS=2048
LLM_TIMEOUT_SECONDS=120
# OCI OpenAI-compatible endpoint
OCI_GENAI_BASE_URL=https://inference.generativeai.us-chicago-1.oci.oraclecloud.com/openai/v1
OCI_GENAI_MODEL=openai.gpt-4.1
OCI_GENAI_API_KEY=sk-ph3FgX6iP3fxAQCXb9IpPIDTadkeeYAWntUWhzcWysIM6zsS
OCI_GENAI_PROJECT_OCID=
# OCI SDK / signer / profiles
OCI_CONFIG_FILE=~/.oci/config
OCI_PROFILE=LATINOAMERICA-Chicago
OCI_COMPARTMENT_ID=ocid1.compartment.oc1..aaaaaaaaexpiw4a7dio64mkfv2t273s2hgdl6mgfvvyv7tycalnjlvpvfl3q
OCI_REGION=us-chicago-1
###############################################################################
# Persistência
###############################################################################
# Opções: memory, autonomous, mongodb
SESSION_REPOSITORY_PROVIDER=sqlite
MEMORY_REPOSITORY_PROVIDER=sqlite
CHECKPOINT_REPOSITORY_PROVIDER=sqlite
SQLITE_DB_PATH=./data/agent_framework.db
# Autonomous Database
ADB_USER=admin
ADB_PASSWORD=Moniquinha1972
ADB_DSN=oradb23ai_high
ADB_WALLET_LOCATION=/mnt/d/Dropbox/ORACLE/LatinoAmerica/Wallet_ORADB23ai
ADB_WALLET_PASSWORD=Moniquinha1972
ADB_TABLE_PREFIX=AGENTFW
# MongoDB - também pode representar Autonomous usando API compatível com Mongo, se habilitada no ambiente
MONGODB_URI=mongodb://mongo:mongopassword@localhost:27017
MONGODB_DATABASE=agent_platform
# Redis
REDIS_URL=redis://localhost:6379/0
ENABLE_REDIS_CACHE=false
###############################################################################
# RAG / Vector / Graph
###############################################################################
VECTOR_STORE_PROVIDER=memory
GRAPH_STORE_PROVIDER=memory
RAG_TOP_K=5
EMBEDDING_PROVIDER=mock
OCI_EMBEDDING_MODEL=cohere.embed-multilingual-v3.0
RAG_FILE_GLOBS=*.md,*.txt,*.yaml,*.yml,*.json
###############################################################################
# Observabilidade
###############################################################################
ENABLE_LANGFUSE=true
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=pk-lf-2f9da109-5b0f-4c78-b61d-9598ed787eba
LANGFUSE_SECRET_KEY=sk-lf-a4cb0cdd-f2ea-4468-9911-cebeb91ba944
LANGFUSE_HOST=http://localhost:3005
ENABLE_OTEL=false
OTEL_EXPORTER_OTLP_ENDPOINT=
OTEL_SERVICE_NAME=ai-agent-template
###############################################################################
# Analytics / Observer corporativo
###############################################################################
# Quando true, AgentObserver publica eventos IC.*, NOC.* e GRL.* nos providers abaixo.
ENABLE_ANALYTICS=false
# Providers aceitos: oci_streaming,pubsub,noop
ANALYTICS_PROVIDERS=pubsub
# Compatibilidade FIRST/TIM: pode informar AGENT_PUBSUB_TOPIC diretamente.
AGENT_PUBSUB_TOPIC=
GCP_PUBSUB_TOPIC_PATH=
GCP_PROJECT_ID=
GCP_PUBSUB_TOPIC=
GCP_PUBSUB_TIMEOUT_SECONDS=30
# Credencial GCP segue padrão Google:
# GOOGLE_APPLICATION_CREDENTIALS=/secrets/gcp-service-account.json
###############################################################################
# OCI Streaming
###############################################################################
ENABLE_OCI_STREAMING=false
OCI_STREAM_ENDPOINT=
OCI_STREAM_OCID=
OCI_STREAM_PARTITION_KEY=agent-events
###############################################################################
# Guardrails, Judges, Supervisor
###############################################################################
ENABLE_INPUT_GUARDRAILS=true
ENABLE_OUTPUT_GUARDRAILS=true
ENABLE_JUDGES=true
ENABLE_SUPERVISOR=true
ENABLE_OUTPUT_SUPERVISOR=true
ENABLE_PARALLEL_GUARDRAILS=true
GUARDRAILS_FAIL_FAST=true
OUTPUT_SUPERVISOR_MAX_RETRIES=3
GUARDRAILS_CONFIG_PATH=./config/guardrails.yaml
JUDGES_CONFIG_PATH=./config/judges.yaml
PROMPT_POLICY_PATH=./config/prompt_policy.yaml
###############################################################################
# Gateway de canais
###############################################################################
DEFAULT_CHANNEL=web
ENABLE_VOICE_ADAPTER=true
ENABLE_WHATSAPP_ADAPTER=true
ENABLE_TEXT_ADAPTER=true
#################################################
# ENTERPRISE ROUTING
#################################################
# Arquivo YAML com intents, keywords, políticas de estado e fallback.
ROUTING_CONFIG_PATH=./config/routing.yaml
# true = usa LLM para classificar quando keywords/estado não resolverem.
# Em produção, costuma ser útil; em desenvolvimento, false evita custo e latência.
ENABLE_LLM_ROUTER=true
###############################################################################
# MCP / Tools
###############################################################################
ENABLE_MCP_TOOLS=true
MCP_SERVERS_CONFIG_PATH=./agent_template_backend/config/mcp_servers.yaml
TOOLS_CONFIG_PATH=./agent_template_backend/config/tools.yaml
MCP_TOOL_TIMEOUT_SECONDS=30
# router = EnterpriseRouter seleciona um agente; supervisor = pode acionar múltiplos agentes
ROUTING_MODE=router
# Usage/cost accounting
USAGE_REPOSITORY_PROVIDER=sqlite
IDENTITY_CONFIG_PATH=./agent_template_backend/config/identity.yaml
MCP_PARAMETER_MAPPING_PATH=./agent_template_backend/config/mcp_parameter_mapping.yaml
# -----------------------------------------------------------------------------
# ConversationSummaryMemory / compressão de contexto conversacional
# -----------------------------------------------------------------------------
ENABLE_CONVERSATION_SUMMARY_MEMORY=true
MEMORY_CONTEXT_STRATEGY=summary
MEMORY_HISTORY_LIMIT=80
MEMORY_RECENT_MESSAGES_LIMIT=8
MEMORY_SUMMARY_TRIGGER_MESSAGES=20
MEMORY_MAX_SUMMARY_CHARS=6000
MEMORY_SUMMARY_USE_LLM=true
MEMORY_INJECT_RECENT_MESSAGES=true
MEMORY_INJECT_SUMMARY=true

View File

@@ -71,6 +71,9 @@ RAG_FILE_GLOBS=*.md,*.txt,*.yaml,*.yml,*.json
# Observabilidade
###############################################################################
ENABLE_LANGFUSE=true
LANGFUSE_TRACE_MODE=verbose # Opcional: verbose, compact
LANGFUSE_ROOT_SPAN_NAME=agent.gateway_message
LANGFUSE_LEGACY_IO_FALLBACK=true
LANGFUSE_PUBLIC_KEY=pk-lf-bd9b0c7e-2b8b-4e5b-a382-284a9b4413b3
LANGFUSE_SECRET_KEY=sk-lf-5f5cc18d-0bb5-424e-b5d0-cb3664d58c20
LANGFUSE_HOST=http://localhost:3005
@@ -85,7 +88,7 @@ ENABLE_LANGFUSE_OPENAI_AUTO_INSTRUMENTATION=true
# Quando true, AgentObserver publica eventos IC.*, NOC.* e GRL.* nos providers abaixo.
ENABLE_ANALYTICS=false
# Providers aceitos: oci_streaming,pubsub,noop
ANALYTICS_PROVIDERS=oci_streaming
ANALYTICS_PROVIDERS=pubsub
# Compatibilidade FIRST/TIM: pode informar AGENT_PUBSUB_TOPIC diretamente.
AGENT_PUBSUB_TOPIC=
GCP_PUBSUB_TOPIC_PATH=

16
.idea/agent_platform_oci.iml generated Normal file
View File

@@ -0,0 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<module type="JAVA_MODULE" version="4">
<component name="NewModuleRootManager" inherit-compiler-output="true">
<exclude-output />
<content url="file://$MODULE_DIR$" />
<orderEntry type="inheritedJdk" />
<orderEntry type="sourceFolder" forTests="false" />
</component>
<component name="TemplatesService">
<option name="TEMPLATE_FOLDERS">
<list>
<option value="$MODULE_DIR$/templates" />
</list>
</option>
</component>
</module>

8
.idea/modules.xml generated Normal file
View File

@@ -0,0 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ProjectModuleManager">
<modules>
<module fileurl="file://$PROJECT_DIR$/.idea/agent_platform_oci.iml" filepath="$PROJECT_DIR$/.idea/agent_platform_oci.iml" />
</modules>
</component>
</project>

6
.idea/vcs.xml generated Normal file
View File

@@ -0,0 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="$PROJECT_DIR$" vcs="Git" />
</component>
</project>

133
.idea/workspace.xml generated
View File

@@ -4,9 +4,8 @@
<option name="autoReloadType" value="SELECTIVE" />
</component>
<component name="ChangeListManager">
<list default="true" id="30a0e1d8-9d7d-469b-b241-f300911cee8a" name="Changes" comment="Ajustes na documentação e remanejamento dos folders">
<list default="true" id="30a0e1d8-9d7d-469b-b241-f300911cee8a" name="Changes" comment="bugfix Alex 2026-06-30">
<change beforePath="$PROJECT_DIR$/.idea/workspace.xml" beforeDir="false" afterPath="$PROJECT_DIR$/.idea/workspace.xml" afterDir="false" />
<change beforePath="$PROJECT_DIR$/MANUAL_AGENT_PLATFORM_GATEWAYS.md" beforeDir="false" afterPath="$PROJECT_DIR$/MANUAL_AGENT_PLATFORM_GATEWAYS.md" afterDir="false" />
</list>
<option name="SHOW_DIALOG" value="false" />
<option name="HIGHLIGHT_CONFLICTS" value="true" />
@@ -31,10 +30,10 @@
&quot;accountId&quot;: &quot;bd799ac0-7624-42b2-999a-b96e6e11f98a&quot;
}
}</component>
<component name="KubernetesApiPersistence"><![CDATA[{}]]></component>
<component name="KubernetesApiProvider"><![CDATA[{
"isMigrated": true
}]]></component>
<component name="KubernetesApiPersistence">{}</component>
<component name="KubernetesApiProvider">{
&quot;isMigrated&quot;: true
}</component>
<component name="PhpWorkspaceProjectConfiguration" interpreter_name="PHP 8.1" />
<component name="ProjectCodeStyleSettingsMigration">
<option name="version" value="2" />
@@ -82,6 +81,10 @@
<updated>1781996088468</updated>
<workItem from="1781996091031" duration="1242000" />
<workItem from="1782038604623" duration="6486000" />
<workItem from="1782047166074" duration="27000" />
<workItem from="1782047194363" duration="961000" />
<workItem from="1782048494672" duration="2790000" />
<workItem from="1782900499027" duration="911000" />
</task>
<task id="LOCAL-00001" summary="Ajustes na documentação e remanejamento dos folders">
<option name="closed" value="true" />
@@ -99,7 +102,119 @@
<option name="project" value="LOCAL" />
<updated>1782045265085</updated>
</task>
<option name="localTasksCounter" value="3" />
<task id="LOCAL-00003" summary="Ajustes na documentação e remanejamento dos folders">
<option name="closed" value="true" />
<created>1782048304579</created>
<option name="number" value="00003" />
<option name="presentableId" value="LOCAL-00003" />
<option name="project" value="LOCAL" />
<updated>1782048304579</updated>
</task>
<task id="LOCAL-00004" summary="Ajustes na documentação e remanejamento dos folders">
<option name="closed" value="true" />
<created>1782048747421</created>
<option name="number" value="00004" />
<option name="presentableId" value="LOCAL-00004" />
<option name="project" value="LOCAL" />
<updated>1782048747421</updated>
</task>
<task id="LOCAL-00005" summary="bugfix Alex 2026-06-24">
<option name="closed" value="true" />
<created>1782321346355</created>
<option name="number" value="00005" />
<option name="presentableId" value="LOCAL-00005" />
<option name="project" value="LOCAL" />
<updated>1782321346355</updated>
</task>
<task id="LOCAL-00006" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900906490</created>
<option name="number" value="00006" />
<option name="presentableId" value="LOCAL-00006" />
<option name="project" value="LOCAL" />
<updated>1782900906490</updated>
</task>
<task id="LOCAL-00007" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900910793</created>
<option name="number" value="00007" />
<option name="presentableId" value="LOCAL-00007" />
<option name="project" value="LOCAL" />
<updated>1782900910793</updated>
</task>
<task id="LOCAL-00008" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900913579</created>
<option name="number" value="00008" />
<option name="presentableId" value="LOCAL-00008" />
<option name="project" value="LOCAL" />
<updated>1782900913579</updated>
</task>
<task id="LOCAL-00009" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900914906</created>
<option name="number" value="00009" />
<option name="presentableId" value="LOCAL-00009" />
<option name="project" value="LOCAL" />
<updated>1782900914906</updated>
</task>
<task id="LOCAL-00010" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900923834</created>
<option name="number" value="00010" />
<option name="presentableId" value="LOCAL-00010" />
<option name="project" value="LOCAL" />
<updated>1782900923834</updated>
</task>
<task id="LOCAL-00011" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900925229</created>
<option name="number" value="00011" />
<option name="presentableId" value="LOCAL-00011" />
<option name="project" value="LOCAL" />
<updated>1782900925229</updated>
</task>
<task id="LOCAL-00012" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900926577</created>
<option name="number" value="00012" />
<option name="presentableId" value="LOCAL-00012" />
<option name="project" value="LOCAL" />
<updated>1782900926577</updated>
</task>
<task id="LOCAL-00013" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900927457</created>
<option name="number" value="00013" />
<option name="presentableId" value="LOCAL-00013" />
<option name="project" value="LOCAL" />
<updated>1782900927457</updated>
</task>
<task id="LOCAL-00014" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900928320</created>
<option name="number" value="00014" />
<option name="presentableId" value="LOCAL-00014" />
<option name="project" value="LOCAL" />
<updated>1782900928320</updated>
</task>
<task id="LOCAL-00015" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900928490</created>
<option name="number" value="00015" />
<option name="presentableId" value="LOCAL-00015" />
<option name="project" value="LOCAL" />
<updated>1782900928490</updated>
</task>
<task id="LOCAL-00016" summary="bugfix Alex 2026-06-30">
<option name="closed" value="true" />
<created>1782900939364</created>
<option name="number" value="00016" />
<option name="presentableId" value="LOCAL-00016" />
<option name="project" value="LOCAL" />
<updated>1782900939364</updated>
</task>
<option name="localTasksCounter" value="17" />
<servers />
</component>
<component name="TypeScriptGeneratedFilesManager">
@@ -107,6 +222,8 @@
</component>
<component name="VcsManagerConfiguration">
<MESSAGE value="Ajustes na documentação e remanejamento dos folders" />
<option name="LAST_COMMIT_MESSAGE" value="Ajustes na documentação e remanejamento dos folders" />
<MESSAGE value="bugfix Alex 2026-06-24" />
<MESSAGE value="bugfix Alex 2026-06-30" />
<option name="LAST_COMMIT_MESSAGE" value="bugfix Alex 2026-06-30" />
</component>
</project>

View File

@@ -0,0 +1,133 @@
# Inventário — Agent Gateway + MCP Gateway Overlay
Este inventário lista os arquivos incluídos no overlay `agent_platform_agent_gateway_mcp_gateway_overlay.zip`, indicando a área, o tipo de alteração e a finalidade de cada arquivo.
## Resumo
| Área | Quantidade |
|---|---:|
| Documentação | 1 |
| Agent Gateway | 10 |
| MCP Gateway | 5 |
| Agent Framework | 4 |
| Template Backend | 2 |
| MCP Server Mock | 2 |
| Deploy | 2 |
## Arquivos por área
### Documentação
| Arquivo | Tipo | Finalidade |
|---|---|---|
| `README_AGENT_GATEWAY_AND_MCP_GATEWAY_EVOLUTION.md` | Novo / overlay | Documento principal do overlay. Explica a nova arquitetura sem AI Gateway separado, com Agent Gateway governando políticas/modelos e MCP Gateway separado para tools. |
### Agent Gateway
| Arquivo | Tipo | Finalidade |
|---|---|---|
| `apps/agent_gateway/app/config/governance_loader.py` | Novo / overlay | Carrega o arquivo YAML de governança do Agent Gateway a partir de AGENT_GATEWAY_GOVERNANCE_CONFIG. |
| `apps/agent_gateway/app/governance/__init__.py` | Novo / overlay | Inicializa o pacote Python de governança do Agent Gateway. |
| `apps/agent_gateway/app/governance/audit.py` | Novo / overlay | Centraliza logging/auditoria das decisões de governança do Agent Gateway, com proteção simples para não logar mensagem completa. |
| `apps/agent_gateway/app/governance/evaluation_hooks.py` | Novo / overlay | Hooks antes e depois da chamada ao backend/runtime. Serve para amostragem, evaluator, scoring ou integração futura com Langfuse. |
| `apps/agent_gateway/app/governance/model_policies.py` | Novo / overlay | Resolve políticas de modelo/profile no Agent Gateway. Define qual provider/model/profile deve ser usado por operação, tenant e agente. |
| `apps/agent_gateway/app/governance/rate_limit.py` | Novo / overlay | Implementa rate limit em memória por tenant, agente e canal antes de encaminhar a requisição ao backend/runtime. |
| `apps/agent_gateway/app/governance/usage.py` | Novo / overlay | Hook para registrar uso de gateway, políticas aplicadas e respostas do backend. Pronto para plugar métricas, banco, Langfuse ou OTEL. |
| `apps/agent_gateway/app/governance_middleware.py` | Novo / overlay | Componente principal de governança do Agent Gateway. Aplica rate limit, resolve model_policy, gera headers/metadados e executa hooks antes/depois do backend. |
| `apps/agent_gateway/app/routes/governed_proxy_example.py` | Novo / overlay | Exemplo de rota governada para demonstrar como aplicar governança antes de encaminhar para o Agent Backend/Runtime. |
| `apps/agent_gateway/config/gateway_governance.yaml` | Novo / overlay | Configuração de governança do Agent Gateway: profiles, operation_profiles, providers permitidos, rate limits, headers propagados e evaluation hooks. |
### MCP Gateway
| Arquivo | Tipo | Finalidade |
|---|---|---|
| `apps/mcp_gateway/Dockerfile` | Novo / overlay | Imagem Docker do MCP Gateway. |
| `apps/mcp_gateway/app/__init__.py` | Novo / overlay | Inicializa o pacote Python da aplicação MCP Gateway. |
| `apps/mcp_gateway/app/main.py` | Novo / overlay | Aplicação FastAPI do MCP Gateway. Expõe health, ready, catálogo de tools e endpoint de invoke com auth, autorização, mapping, cache, timeout e retry. |
| `apps/mcp_gateway/config/mcp_gateway.yaml` | Novo / overlay | Configuração central do MCP Gateway: MCP servers, tools, versões, cache, timeout, retry, autorização por agente/canal e mapping BusinessContext → parâmetros. |
| `apps/mcp_gateway/requirements.txt` | Novo / overlay | Dependências Python do MCP Gateway. |
### Agent Framework
| Arquivo | Tipo | Finalidade |
|---|---|---|
| `libs/agent_framework/src/agent_framework/gateway_policy_context.py` | Novo / overlay | Helper no framework para o Runtime ler a política de modelo enviada pelo Agent Gateway em state['metadata']['model_policy']. |
| `libs/agent_framework/src/agent_framework/gateways/__init__.py` | Novo / overlay | Inicializa o pacote de clients de gateways no framework, exportando MCPGatewayClient. |
| `libs/agent_framework/src/agent_framework/gateways/mcp_gateway_client.py` | Novo / overlay | Client assíncrono do framework para chamar o MCP Gateway: listar tools e executar tools. |
| `libs/agent_framework/src/agent_framework/runtime_mcp_gateway_adapter.py` | Novo / overlay | Mixin opcional para agentes/runtime chamarem tools via MCP Gateway e anexarem resultados em state['mcp_results']. |
### Template Backend
| Arquivo | Tipo | Finalidade |
|---|---|---|
| `templates/agent_template_backend/app/mcp_gateway_client_factory.py` | Novo / overlay | Factory no template backend para construir MCPGatewayClient a partir de variáveis de ambiente. |
### MCP Server Mock
| Arquivo | Tipo | Finalidade |
|---|---|---|
| `mcp/servers/mock_telecom_mcp/app.py` | Novo / overlay | Mock MCP Server com tools consultar_fatura e consultar_pagamentos para validar o MCP Gateway localmente. |
| `mcp/servers/mock_telecom_mcp/requirements.txt` | Novo / overlay | Dependências do mock MCP Server de telecom usado para testes locais. |
### Deploy
| Arquivo | Tipo | Finalidade |
|---|---|---|
| `deploy/docker/docker-compose.mcp-gateway.yml` | Novo / overlay | Docker Compose para subir MCP Gateway e mock_telecom_mcp localmente. |
| `deploy/k8s/mcp-gateway.yaml` | Novo / overlay | Manifest Kubernetes de Deployment e Service do MCP Gateway. |
## Observações de integração
### Agent Gateway
Os arquivos em `apps/agent_gateway` não criam um novo serviço. Eles evoluem o Agent Gateway existente para atuar como gateway dedicado da plataforma, centralizando:
- políticas de modelo/profile;
- rate limit;
- auditoria;
- hooks de avaliação;
- propagação de metadados de governança para o Runtime.
A rota `governed_proxy_example.py` é um exemplo de integração. O handler real do `POST /gateway/message` deve aplicar:
```python
governed_body, headers = governance.prepare_backend_request(body)
```
antes de chamar o backend/runtime, e:
```python
return governance.process_backend_response(data)
```
após receber a resposta.
### MCP Gateway
O MCP Gateway é um serviço separado. Ele centraliza:
- catálogo de tools;
- autorização por agente/canal;
- versionamento de tools;
- mapping de BusinessContext para parâmetros;
- cache;
- timeout;
- retry;
- auditoria simples.
### Runtime / Backend
O Runtime continua responsável por:
- LangGraph;
- estado;
- memória;
- checkpoints;
- fluxo;
- providers LLM existentes.
O Runtime passa a chamar tools via MCP Gateway usando `MCPGatewayClient` e/ou `MCPGatewayRuntimeMixin`.
### AI Gateway
Este overlay não cria `apps/ai_gateway`. A governança de modelo fica no Agent Gateway, e a execução LLM continua no Runtime/backend usando os providers já existentes.

View File

@@ -0,0 +1,627 @@
# Manual de Execução Local
## Agent Gateway + MCP Gateway + Agent Template Backend + Frontend
## 1. Arquitetura de execução
A arquitetura local fica assim:
```text
Frontend
porta 5173
Agent Gateway
porta 9000
Agent Template Backend / Agent Runtime
porta 8000
MCP Gateway
porta 8300
MCP Server / Mock Telecom MCP
porta 8001
```
A governança de modelo, rate limit, auditoria e políticas ficam no **Agent Gateway**.
O **Agent Runtime / Agent Template Backend** continua responsável por:
- LangGraph;
- estado;
- memória;
- checkpoints;
- supervisor/router;
- guardrails;
- judges;
- chamada LLM via providers existentes;
- chamada de tools via MCP Gateway.
---
## 2. Portas
| Componente | Porta | URL |
|---|---:|---|
| Frontend | 5173 | `http://localhost:5173` |
| Agent Gateway | 9000 | `http://localhost:9000` |
| Agent Template Backend | 8000 | `http://localhost:8000` |
| MCP Gateway | 8300 | `http://localhost:8300` |
| MCP Server / Mock Telecom MCP | 8001 | `http://localhost:8001` |
---
## 3. Ordem recomendada para subir
Subir nesta ordem:
1. MCP Server / Mock Telecom MCP
2. MCP Gateway
3. Agent Template Backend
4. Agent Gateway
5. Frontend
---
# 4. Terminal 1 — MCP Server / Mock Telecom MCP
Se estiver usando o mock incluído no overlay:
```bash
cd agent_platform_oci/mcp/servers/mock_telecom_mcp
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
uvicorn app:app --host 0.0.0.0 --port 8001 --reload
```
Validar:
```bash
curl http://localhost:8001/health
```
Resultado esperado:
```json
{
"status": "ok",
"service": "mock_telecom_mcp"
}
```
---
# 5. Terminal 2 — MCP Gateway
```bash
cd agent_platform_oci/apps/mcp_gateway
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
export MCP_GATEWAY_CONFIG_PATH=config/mcp_gateway.yaml
uvicorn app.main:app --host 0.0.0.0 --port 8300 --reload
```
Validar health:
```bash
curl http://localhost:8300/health
```
Validar readiness:
```bash
curl http://localhost:8300/ready
```
Listar tools:
```bash
curl -s http://localhost:8300/v1/tools | jq
```
Executar tool:
```bash
curl -s -X POST http://localhost:8300/v1/tools/consultar_fatura/invoke \
-H "Content-Type: application/json" \
-d '{
"tenant_id": "default",
"agent_id": "telecom_contas",
"channel": "web",
"tool_name": "consultar_fatura",
"business_context": {
"customer_key": "11999999999",
"contract_key": "INV-001",
"session_key": "session-001"
}
}' | jq
```
Resultado esperado:
```json
{
"tool_name": "consultar_fatura",
"version": "1.0.0",
"ok": true,
"data": {
"invoice_id": "INV-001",
"msisdn": "11999999999",
"valor_total": 249.9,
"vencimento": "2026-06-10",
"status": "ABERTA"
}
}
```
---
# 6. Terminal 3 — Agent Template Backend / Agent Runtime
```bash
cd agent_platform_oci/templates/agent_template_backend
```
ou, se o seu backend estiver em outra pasta:
```bash
cd agent_platform_oci/templates/agent_template_backend
```
Ativar ambiente:
```bash
source .venv/bin/activate
```
Se ainda não existir `.venv`:
```bash
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```
Configurar variáveis:
```bash
export MCP_GATEWAY_ENABLED=true
export MCP_GATEWAY_URL=http://localhost:8300
export MCP_GATEWAY_TIMEOUT_SECONDS=60
export AGENT_GATEWAY_GOVERNANCE_CONFIG=config/gateway_governance.yaml
```
Se estiver usando OCI/OpenAI-compatible, manter também as variáveis já existentes do backend:
```bash
export LLM_PROVIDER=oci_openai
export OCI_GENAI_API_KEY=<sua-chave>
```
ou, para mock:
```bash
export LLM_PROVIDER=mock
```
Subir backend:
```bash
python -m uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
```
Validar:
```bash
curl http://localhost:8000/health
```
Validar agentes:
```bash
curl http://localhost:8000/agents | jq
```
Testar backend direto:
```bash
curl -s -X POST http://localhost:8000/gateway/message \
-H "Content-Type: application/json" \
-d '{
"channel": "web",
"tenant_id": "default",
"agent_id": "telecom_contas",
"payload": {
"message": "Quero consultar minha fatura",
"session_id": "session-001",
"user_id": "user-001",
"message_id": "msg-001",
"business_context": {
"customer_key": "11999999999",
"contract_key": "INV-001",
"session_key": "session-001"
}
}
}' | jq
```
---
# 7. Terminal 4 — Agent Gateway
```bash
cd agent_platform_oci/apps/agent_gateway
```
Ativar ambiente:
```bash
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```
Configurar variáveis:
```bash
export DEFAULT_AGENT_BACKEND_URL=http://localhost:8000
export AGENT_GATEWAY_GOVERNANCE_CONFIG=config/gateway_governance.yaml
```
Subir Agent Gateway:
```bash
python -m uvicorn app.main:app --host 0.0.0.0 --port 9000 --reload
```
Validar:
```bash
curl http://localhost:9000/health
```
Se a rota governada de exemplo estiver registrada no `app.main`, testar:
```bash
curl -s -X POST http://localhost:9000/gateway/message/governed \
-H "Content-Type: application/json" \
-d '{
"channel": "web",
"tenant_id": "default",
"agent_id": "telecom_contas",
"payload": {
"message": "Quero consultar minha fatura",
"session_id": "session-001",
"user_id": "user-001",
"message_id": "msg-001",
"metadata": {
"operation": "agent.final_answer"
},
"business_context": {
"customer_key": "11999999999",
"contract_key": "INV-001",
"session_key": "session-001"
}
}
}' | jq
```
Se a rota real for `/gateway/message`, testar:
```bash
curl -s -X POST http://localhost:9000/gateway/message \
-H "Content-Type: application/json" \
-d '{
"channel": "web",
"tenant_id": "default",
"agent_id": "telecom_contas",
"payload": {
"message": "Quero consultar minha fatura",
"session_id": "session-001",
"user_id": "user-001",
"message_id": "msg-001",
"metadata": {
"operation": "agent.final_answer"
},
"business_context": {
"customer_key": "11999999999",
"contract_key": "INV-001",
"session_key": "session-001"
}
}
}' | jq
```
---
# 8. Terminal 5 — Frontend
```bash
cd agent_platform_oci/agent_frontend
```
ou a pasta onde estiver o frontend.
Instalar dependências:
```bash
npm install
```
Subir:
```bash
npm run dev -- --host 0.0.0.0 --port 5173
```
Abrir:
```text
http://localhost:5173
```
Configurar no frontend:
```text
Backend URL: http://localhost:9000
Agent: telecom_contas
Session ID: session-001
Customer Key: 11999999999
Contract Key: INV-001
```
O frontend deve chamar o **Agent Gateway** na porta 9000, não o MCP Gateway.
---
# 9. Fluxo final esperado
```text
Frontend 5173
Agent Gateway 9000
Agent Template Backend 8000
MCP Gateway 8300
Mock Telecom MCP 8001
```
---
# 10. Docker Compose para MCP Gateway + Mock MCP
Também é possível subir MCP Gateway + Mock MCP com Docker Compose:
```bash
cd agent_platform_oci
docker compose -f deploy/docker/docker-compose.mcp-gateway.yml up --build
```
Isso sobe:
```text
MCP Gateway http://localhost:8300
Mock Telecom MCP http://localhost:8001
```
Depois subir manualmente:
- Agent Template Backend na porta 8000;
- Agent Gateway na porta 9000;
- Frontend na porta 5173.
---
# 11. Checklist de validação
## MCP Server
```bash
curl http://localhost:8001/health
```
## MCP Gateway
```bash
curl http://localhost:8300/health
curl http://localhost:8300/v1/tools
```
## Backend Runtime
```bash
curl http://localhost:8000/health
curl http://localhost:8000/agents
```
## Agent Gateway
```bash
curl http://localhost:9000/health
```
## Frontend
```text
http://localhost:5173
```
---
# 12. Erros comuns
## 12.1. Frontend chamando porta errada
Errado:
```text
Frontend → http://localhost:8000
```
Correto:
```text
Frontend → http://localhost:9000
```
Se você quiser testar sem Agent Gateway, pode apontar temporariamente para 8000. Mas no modelo final, o frontend deve usar o Agent Gateway.
---
## 12.2. MCP Gateway sem MCP Server
Sintoma:
```text
MCP server unavailable
```
Correção:
```bash
curl http://localhost:8001/health
```
Se falhar, subir o mock MCP server.
---
## 12.3. Tool sem BusinessContext
Sintoma:
```json
{
"missing_business_keys": ["customer_key", "contract_key"]
}
```
Correção:
enviar:
```json
"business_context": {
"customer_key": "11999999999",
"contract_key": "INV-001",
"session_key": "session-001"
}
```
---
## 12.4. Agent Gateway não encontra backend
Sintoma:
```text
Connection refused http://localhost:8000
```
Correção:
validar:
```bash
curl http://localhost:8000/health
```
e configurar:
```bash
export DEFAULT_AGENT_BACKEND_URL=http://localhost:8000
```
---
## 12.5. Rota governada não registrada
Se `/gateway/message/governed` retornar 404, significa que o arquivo de exemplo ainda não foi incluído no `app.main`.
Nesse caso, use a rota real `/gateway/message` ou registre no `main.py`:
```python
from app.routes.governed_proxy_example import router as governed_router
app.include_router(governed_router)
```
---
# 13. Variáveis consolidadas
## Agent Gateway
```env
DEFAULT_AGENT_BACKEND_URL=http://localhost:8000
AGENT_GATEWAY_GOVERNANCE_CONFIG=config/gateway_governance.yaml
```
## Agent Template Backend
```env
MCP_GATEWAY_ENABLED=true
MCP_GATEWAY_URL=http://localhost:8300
MCP_GATEWAY_TIMEOUT_SECONDS=60
LLM_PROVIDER=mock
```
## MCP Gateway
```env
MCP_GATEWAY_CONFIG_PATH=config/mcp_gateway.yaml
```
---
# 14. Resumo rápido
Em cinco terminais:
```bash
# Terminal 1
cd mcp/servers/mock_telecom_mcp
source .venv/bin/activate
uvicorn app:app --host 0.0.0.0 --port 8001 --reload
# Terminal 2
cd apps/mcp_gateway
source .venv/bin/activate
export MCP_GATEWAY_CONFIG_PATH=config/mcp_gateway.yaml
uvicorn app.main:app --host 0.0.0.0 --port 8300 --reload
# Terminal 3
cd templates/agent_template_backend
source .venv/bin/activate
export MCP_GATEWAY_ENABLED=true
export MCP_GATEWAY_URL=http://localhost:8300
uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
# Terminal 4
cd apps/agent_gateway
source .venv/bin/activate
export DEFAULT_AGENT_BACKEND_URL=http://localhost:8000
export AGENT_GATEWAY_GOVERNANCE_CONFIG=config/gateway_governance.yaml
uvicorn app.main:app --host 0.0.0.0 --port 9000 --reload
# Terminal 5
cd agent_frontend
npm install
npm run dev -- --host 0.0.0.0 --port 5173
```

View File

@@ -0,0 +1,93 @@
# MCP Gateway Runbook
## Arquitetura corrigida
O backend/agente não deve chamar diretamente os MCP servers finais. O fluxo correto é:
```text
agent_template_backend / agent_framework
-> MCP Gateway Client
-> apps/mcp_gateway
-> mcp/servers/telecom_mcp_server ou mcp/servers/retail_mcp_server
```
## Subir localmente
A partir da raiz do projeto:
### Terminal 1 - Telecom MCP Server
```bash
cd mcp/servers/telecom_mcp_server
python -m uvicorn main:app --host 0.0.0.0 --port 8100 --reload
```
### Terminal 2 - Retail MCP Server
```bash
cd mcp/servers/retail_mcp_server
python -m uvicorn main:app --host 0.0.0.0 --port 8200 --reload
```
### Terminal 3 - MCP Gateway
```bash
cd apps/mcp_gateway
export MCP_GATEWAY_CONFIG_PATH=config/mcp_gateway.yaml
python -m uvicorn app.main:app --host 0.0.0.0 --port 8300 --reload
```
### Terminal 4 - Backend/agente
No `.env` do backend/agente ou do runtime que usa o `agent_framework`, habilite:
```env
ENABLE_MCP_TOOLS=true
MCP_GATEWAY_ENABLED=true
MCP_GATEWAY_URL=http://localhost:8300
MCP_GATEWAY_AGENT_ID=telecom_contas
MCP_GATEWAY_TENANT_ID=default
```
## Testes rápidos
### Health do gateway
```bash
curl http://localhost:8300/health
```
### Lista de tools expostas pelo gateway
```bash
curl http://localhost:8300/v1/tools
```
### Chamada de tool via gateway
```bash
curl -X POST http://localhost:8300/v1/tools/consultar_fatura/invoke \
-H 'Content-Type: application/json' \
-d '{
"tenant_id": "default",
"agent_id": "telecom_contas",
"channel": "web",
"tool_name": "consultar_fatura",
"arguments": {
"msisdn": "11999999999",
"invoice_id": "INV-123"
},
"business_context": {},
"metadata": {"session_id": "local-test"}
}'
```
Resposta esperada: `ok: true`, `data.invoice_id`, `data.msisdn`, `metadata.server: telecom`.
## O que foi corrigido
- `apps/mcp_gateway/config/mcp_gateway.yaml` agora aponta para os MCP servers reais nas portas `8100` e `8200`.
- O MCP Gateway agora suporta o contrato legado dos MCP servers: `POST /mcp/tools/call` com `{tool_name, arguments}`.
- O `agent_framework` ganhou flags `MCP_GATEWAY_ENABLED`, `MCP_GATEWAY_URL`, `MCP_GATEWAY_TOKEN`, `MCP_GATEWAY_AGENT_ID` e `MCP_GATEWAY_TENANT_ID`.
- O `MCPToolRouter` passa a chamar o MCP Gateway quando `MCP_GATEWAY_ENABLED=true`.
- `libs/agent_framework/config/mcp_servers.yaml` foi mantido como registry lógico/fallback, não como caminho principal quando o gateway está ativo.

View File

@@ -0,0 +1,129 @@
# Agent Platform OCI — Agent Gateway + MCP Gateway Evolution
Este overlay remove o conceito de `AI Gateway` separado.
## Arquitetura
```text
Frontend
Agent Gateway
├── governance
├── model policies
├── rate limit
├── audit
└── evaluation hooks
Agent Backend / Runtime
├── LangGraph
├── state
├── memory
├── checkpoints
└── LLM providers via profiles existentes
MCP Gateway
MCP Servers
```
## O que entra no Agent Gateway
```text
apps/agent_gateway/app/governance/
apps/agent_gateway/app/governance_middleware.py
apps/agent_gateway/app/routes/governed_proxy_example.py
apps/agent_gateway/config/gateway_governance.yaml
```
## O que entra no MCP Gateway
```text
apps/mcp_gateway/
libs/agent_framework/src/agent_framework/gateways/mcp_gateway_client.py
libs/agent_framework/src/agent_framework/runtime_mcp_gateway_adapter.py
```
## Aplicar overlay
```bash
unzip agent_platform_agent_gateway_mcp_gateway_overlay.zip -d /tmp/overlay
rsync -av /tmp/overlay/ ./
```
## Subir MCP Gateway local
```bash
docker compose -f deploy/docker/docker-compose.mcp-gateway.yml up --build
```
Serviços:
```text
MCP Gateway http://localhost:8300
Mock Telecom MCP http://localhost:8001
```
## Testar MCP Gateway
```bash
curl http://localhost:8300/health
curl http://localhost:8300/v1/tools
```
Executar tool:
```bash
curl -s -X POST http://localhost:8300/v1/tools/consultar_fatura/invoke \
-H "Content-Type: application/json" \
-d '{
"tenant_id": "default",
"agent_id": "telecom_contas",
"channel": "web",
"tool_name": "consultar_fatura",
"business_context": {
"customer_key": "11999999999",
"contract_key": "INV-001",
"session_key": "session-001"
}
}' | jq
```
## Como plugar no Agent Gateway
No handler real do `POST /gateway/message`, antes de encaminhar ao backend/runtime:
```python
governed_body, headers = governance.prepare_backend_request(body)
```
Ao receber resposta do backend:
```python
return governance.process_backend_response(data)
```
O arquivo abaixo mostra um exemplo completo:
```text
apps/agent_gateway/app/routes/governed_proxy_example.py
```
## Variáveis do Runtime
```env
MCP_GATEWAY_ENABLED=true
MCP_GATEWAY_URL=http://localhost:8300
MCP_GATEWAY_TIMEOUT_SECONDS=60
```
## Importante
Não existe `apps/ai_gateway`.
A governança de modelo fica no Agent Gateway como policy/metadados.
O Runtime continua usando os LLM providers existentes, podendo ler a política enviada pelo Gateway em:
```python
state["metadata"]["model_policy"]
```

View File

@@ -33,7 +33,7 @@ Define a arquitetura geral da plataforma, seus componentes principais, estrutura
Descreve o runtime de execução conversacional dos agentes, incluindo LangGraph, estado, memória, checkpoints, roteamento, supervisor, BusinessContext, integração com MCP, RAG, eventos e tratamento de erros.
### [SPEC-003 — AI Gateway](specs/SPEC-003-AI-Gateway.md)
### [SPEC-003 — Agent Gateway](specs/SPEC-003-Agent-Gateway.md)
Especifica o gateway responsável por centralizar chamadas de LLM e embeddings, incluindo contratos de request/response, profiles, providers, autenticação OCI, fallback, rate limit, métricas, segurança e observabilidade.
@@ -342,6 +342,7 @@ RAG_TOP_K=5
EMBEDDING_PROVIDER=mock
ENABLE_LANGFUSE=false
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_HOST=http://localhost:3005
ENABLE_OTEL=false
OTEL_SERVICE_NAME=ai-agent-template
@@ -414,6 +415,7 @@ Para usar Langfuse:
```env
ENABLE_LANGFUSE=true
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=<public-key>
LANGFUSE_SECRET_KEY=<secret-key>
LANGFUSE_HOST=http://localhost:3005
@@ -8553,6 +8555,7 @@ Verifique:
```env
ENABLE_LANGFUSE=true
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=<public-key>
LANGFUSE_SECRET_KEY=<secret-key>
LANGFUSE_HOST=http://localhost:3005

View File

@@ -32,7 +32,7 @@ Defines the overall platform architecture, its main components, repository struc
Describes the conversational execution runtime for agents, including LangGraph, state management, memory, checkpoints, routing, supervisor, BusinessContext, MCP integration, RAG, events, and error handling.
### [SPEC-003 — AI Gateway](specs/SPEC-003-AI-Gateway.md)
### [SPEC-003 — Agent Gateway](specs/SPEC-003-Agent-Gateway.md)
Specifies the gateway responsible for centralizing LLM and embedding calls, including request/response contracts, profiles, providers, OCI authentication, fallback, rate limiting, metrics, security, and observability.
@@ -340,6 +340,7 @@ RAG_TOP_K=5
EMBEDDING_PROVIDER=mock
ENABLE_LANGFUSE=false
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_HOST=http://localhost:3005
ENABLE_OTEL=false
OTEL_SERVICE_NAME=ai-agent-template
@@ -412,6 +413,7 @@ To use Langfuse:
```env
ENABLE_LANGFUSE=true
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=<public-key>
LANGFUSE_SECRET_KEY=<secret-key>
LANGFUSE_HOST=http://localhost:3005
@@ -8461,6 +8463,7 @@ Check:
```env
ENABLE_LANGFUSE=true
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=<public-key>
LANGFUSE_SECRET_KEY=<secret-key>
LANGFUSE_HOST=http://localhost:3005

View File

@@ -45,14 +45,25 @@ class ToolResult(BaseModel):
metadata: dict[str, Any] = Field(default_factory=dict)
class DiscoveryResult(BaseModel):
ok: bool
servers_scanned: int = 0
tools_discovered: int = 0
errors: list[dict[str, Any]] = Field(default_factory=list)
tools: list[str] = Field(default_factory=list)
def load_config() -> dict[str, Any]:
path = Path(os.getenv("MCP_GATEWAY_CONFIG_PATH", "config/mcp_gateway.yaml"))
return yaml.safe_load(path.read_text(encoding="utf-8")) or {}
config = load_config()
static_tools: dict[str, dict[str, Any]] = dict(config.get("tools") or {})
discovered_tools: dict[str, dict[str, Any]] = {}
discovery_state: dict[str, Any] = {"last_sync": None, "errors": [], "tools": []}
cache: dict[str, tuple[float, Any]] = {}
app = FastAPI(title="Agent Platform OCI - MCP Gateway", version="1.0.0")
app = FastAPI(title="Agent Platform OCI - MCP Gateway", version="1.1.0")
def audit(name: str, payload: dict[str, Any]) -> None:
@@ -70,6 +81,13 @@ def auth_check(authorization: str | None) -> None:
raise HTTPException(status_code=403, detail="Invalid MCP Gateway token")
def all_tools() -> dict[str, dict[str, Any]]:
merged = dict(discovered_tools)
# Static config wins over discovery so operators can override metadata safely.
merged.update(static_tools)
return merged
def map_arguments(tool_name: str, args: dict[str, Any], bc: dict[str, Any]) -> dict[str, Any]:
result = dict(args or {})
for source, target in ((config.get("parameter_mapping") or {}).get(tool_name) or {}).items():
@@ -100,16 +118,22 @@ async def post_with_retry(url: str, payload: dict[str, Any], timeout: int, retry
raise RuntimeError(str(last_exc))
def build_server_payload(tool_name: str, args: dict[str, Any], server: dict[str, Any], tool: dict[str, Any]) -> dict[str, Any]:
"""Builds the payload expected by the downstream MCP server.
async def get_json(url: str, timeout: int) -> Any:
async with httpx.AsyncClient(timeout=timeout) as client:
resp = await client.get(url)
resp.raise_for_status()
return resp.json()
The example servers under mcp/servers expose the framework legacy contract:
POST /mcp/tools/call {tool_name, arguments}. Some mocks expose direct
tool endpoints that accept only the argument object. The gateway supports
both shapes through the server/tool config.
def build_server_payload(tool_name: str, args: dict[str, Any], server: dict[str, Any], tool: dict[str, Any]) -> dict[str, Any]:
"""Build the payload expected by the downstream MCP server.
Supported protocols:
- legacy_http/framework_http: POST /mcp/tools/call with {tool_name, arguments}
- direct_http: POST to the configured endpoint with only the argument object
"""
protocol = str(tool.get("protocol") or server.get("protocol") or "legacy_http")
if protocol in {"legacy_http", "framework_http"}:
if protocol in {"legacy_http", "framework_http", "fastmcp_http"}:
return {"tool_name": tool_name, "arguments": args or {}}
return args or {}
@@ -121,36 +145,202 @@ def normalize_server_response(data: Any) -> tuple[bool, Any, str | None, dict[st
return True, data, None, {}
def _tool_name(raw: dict[str, Any]) -> str | None:
return raw.get("name") or raw.get("tool_name") or raw.get("id")
def _tool_schema(raw: dict[str, Any]) -> dict[str, Any]:
schema = raw.get("input_schema") or raw.get("inputSchema") or raw.get("schema") or {}
if isinstance(schema, dict):
return schema
return {}
def _extract_tools_from_catalog(catalog: Any) -> list[dict[str, Any]]:
"""Normalize common MCP/FastMCP/custom catalog shapes.
Accepted examples:
- {"tools": [{"name": "x", "description": "...", "input_schema": {...}}]}
- [{"name": "x", "inputSchema": {...}}]
- {"server_id": "s", "capabilities": {"tools": [...]}}
- {"data": {"tools": [...]}}
"""
if isinstance(catalog, list):
return [x for x in catalog if isinstance(x, dict)]
if not isinstance(catalog, dict):
return []
if isinstance(catalog.get("tools"), list):
return [x for x in catalog["tools"] if isinstance(x, dict)]
data = catalog.get("data")
if isinstance(data, dict) and isinstance(data.get("tools"), list):
return [x for x in data["tools"] if isinstance(x, dict)]
caps = catalog.get("capabilities")
if isinstance(caps, dict) and isinstance(caps.get("tools"), list):
return [x for x in caps["tools"] if isinstance(x, dict)]
return []
def _catalog_urls(server_id: str, server: dict[str, Any]) -> list[str]:
if server.get("manifest_url"):
return [str(server["manifest_url"])]
base = str(server.get("url", "")).rstrip("/")
if server.get("catalog_endpoint"):
return [f"{base}{server['catalog_endpoint']}"]
discovery = config.get("discovery") or {}
endpoints = server.get("discovery_endpoints") or discovery.get("default_catalog_endpoints") or [
"/.well-known/mcp-server.json",
"/manifest",
"/mcp/tools",
"/tools",
"/v1/tools",
]
return [f"{base}{endpoint}" for endpoint in endpoints]
def _endpoint_for_tool(server: dict[str, Any], raw_tool: dict[str, Any]) -> str:
if raw_tool.get("endpoint"):
return str(raw_tool["endpoint"])
protocol = str(raw_tool.get("protocol") or server.get("protocol") or "legacy_http")
if protocol in {"legacy_http", "framework_http", "fastmcp_http"}:
return str(server.get("invoke_endpoint") or "/tools/call")
name = _tool_name(raw_tool) or ""
return str(server.get("invoke_endpoint") or f"/tools/{name}")
def normalize_discovered_tool(server_id: str, server: dict[str, Any], raw_tool: dict[str, Any]) -> tuple[str, dict[str, Any]] | None:
name = _tool_name(raw_tool)
if not name:
return None
discovery = config.get("discovery") or {}
defaults = discovery.get("tool_defaults") or {}
tool_cfg: dict[str, Any] = {
"version": str(raw_tool.get("version") or defaults.get("version") or "1.0.0"),
"server": server_id,
"endpoint": _endpoint_for_tool(server, raw_tool),
"protocol": raw_tool.get("protocol") or server.get("protocol") or defaults.get("protocol") or "legacy_http",
"enabled": bool(raw_tool.get("enabled", defaults.get("enabled", True))),
"idempotent": bool(raw_tool.get("idempotent", defaults.get("idempotent", True))),
"cache_ttl_seconds": int(raw_tool.get("cache_ttl_seconds", defaults.get("cache_ttl_seconds", 0)) or 0),
"timeout_seconds": int(raw_tool.get("timeout_seconds", server.get("timeout_seconds", defaults.get("timeout_seconds", 30))) or 30),
"retry": raw_tool.get("retry") or defaults.get("retry") or {"enabled": False},
"allowed_agents": raw_tool.get("allowed_agents") or defaults.get("allowed_agents") or [],
"allowed_channels": raw_tool.get("allowed_channels") or defaults.get("allowed_channels") or [],
"required_business_keys": raw_tool.get("required_business_keys") or defaults.get("required_business_keys") or [],
"description": raw_tool.get("description") or raw_tool.get("doc") or "",
"input_schema": _tool_schema(raw_tool),
"source": "discovery",
}
return name, tool_cfg
async def discover_server(server_id: str, server: dict[str, Any]) -> tuple[list[tuple[str, dict[str, Any]]], list[dict[str, Any]]]:
errors: list[dict[str, Any]] = []
if not server.get("enabled", True):
return [], []
if not server.get("discover", False):
return [], []
timeout = int((config.get("discovery") or {}).get("timeout_seconds", server.get("timeout_seconds", 10)) or 10)
for url in _catalog_urls(server_id, server):
try:
catalog = await get_json(url, timeout=timeout)
raw_tools = _extract_tools_from_catalog(catalog)
normalized = []
for raw in raw_tools:
item = normalize_discovered_tool(server_id, server, raw)
if item:
normalized.append(item)
if normalized:
return normalized, errors
errors.append({"server": server_id, "url": url, "error": "catalog returned no tools"})
except Exception as exc:
errors.append({"server": server_id, "url": url, "error": str(exc)})
return [], errors
async def sync_discovery() -> DiscoveryResult:
discovered_tools.clear()
errors: list[dict[str, Any]] = []
tools_count = 0
scanned = 0
for server_id, server in (config.get("servers") or {}).items():
if not isinstance(server, dict) or not server.get("discover", False):
continue
scanned += 1
normalized, server_errors = await discover_server(server_id, server)
errors.extend(server_errors)
for name, tool_cfg in normalized:
discovered_tools[name] = tool_cfg
tools_count += 1
discovery_state.update({
"last_sync": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"errors": errors,
"tools": sorted(discovered_tools.keys()),
})
audit("mcp.discovery.completed", {"servers_scanned": scanned, "tools_discovered": tools_count, "errors": len(errors)})
return DiscoveryResult(ok=not errors, servers_scanned=scanned, tools_discovered=tools_count, errors=errors, tools=sorted(discovered_tools.keys()))
@app.on_event("startup")
async def startup_discovery() -> None:
discovery = config.get("discovery") or {}
if discovery.get("enabled", False) and discovery.get("sync_on_startup", True):
try:
await sync_discovery()
except Exception as exc:
audit("mcp.discovery.failed", {"error": str(exc)})
@app.get("/health")
async def health():
return {"status": "ok", "service": "mcp_gateway"}
return {"status": "ok", "service": "mcp_gateway", "version": "1.1.0"}
@app.get("/ready")
async def ready():
enabled_tools = [k for k, v in (config.get("tools") or {}).items() if v.get("enabled", True)]
return {"status": "ready", "tools": enabled_tools}
enabled_tools = [k for k, v in all_tools().items() if v.get("enabled", True)]
return {"status": "ready", "tools": enabled_tools, "discovery": discovery_state}
@app.get("/v1/tools")
async def tools():
return {"tools": [{"name": name, **cfg} for name, cfg in (config.get("tools") or {}).items()]}
return {"tools": [{"name": name, **cfg} for name, cfg in all_tools().items()]}
@app.get("/v1/tools/{tool_name}")
async def tool_detail(tool_name: str):
tool = (config.get("tools") or {}).get(tool_name)
tool = all_tools().get(tool_name)
if not tool:
raise HTTPException(status_code=404, detail=f"Tool not found: {tool_name}")
return {"name": tool_name, **tool}
@app.get("/v1/discovery/servers")
async def discovery_servers():
servers = []
for server_id, server in (config.get("servers") or {}).items():
servers.append({
"id": server_id,
"enabled": server.get("enabled", True),
"discover": server.get("discover", False),
"url": server.get("url"),
"manifest_url": server.get("manifest_url"),
"catalog_endpoint": server.get("catalog_endpoint"),
})
return {"servers": servers, "state": discovery_state}
@app.post("/v1/discovery/sync", response_model=DiscoveryResult)
async def discovery_sync(authorization: str | None = Header(default=None)):
auth_check(authorization)
return await sync_discovery()
@app.post("/v1/tools/{tool_name}/invoke", response_model=ToolResult)
async def invoke(tool_name: str, invocation: ToolInvocation, authorization: str | None = Header(default=None)):
started = time.perf_counter()
auth_check(authorization)
tool = (config.get("tools") or {}).get(tool_name)
tool = all_tools().get(tool_name)
if not tool or not tool.get("enabled", True):
raise HTTPException(status_code=404, detail=f"Tool not found or disabled: {tool_name}")
@@ -217,7 +407,7 @@ async def invoke(tool_name: str, invocation: ToolInvocation, authorization: str
error=error,
cache={"hit": False, "key": ck, "ttl_seconds": ttl},
latency_ms=latency_ms,
metadata={"server": tool.get("server"), **server_metadata},
metadata={"server": tool.get("server"), "source": tool.get("source", "static"), **server_metadata},
)
except Exception as exc:
latency_ms = int((time.perf_counter() - started) * 1000)
@@ -228,5 +418,5 @@ async def invoke(tool_name: str, invocation: ToolInvocation, authorization: str
ok=False,
error=str(exc),
latency_ms=latency_ms,
metadata={"server": tool.get("server")},
metadata={"server": tool.get("server"), "source": tool.get("source", "static")},
)

View File

@@ -1,9 +1,36 @@
# Dedicated MCP Gateway configuration.
# The agent backend/framework calls this gateway; this gateway calls the final MCP servers.
# Discovery allows the gateway to sync tool catalogs from registered MCP servers.
# Static tools below remain supported and override discovered tools with the same name.
discovery:
enabled: true
sync_on_startup: true
timeout_seconds: 10
default_catalog_endpoints:
- /.well-known/mcp-server.json
- /manifest
- /mcp/tools
- /tools/list
- /tools
- /v1/tools
tool_defaults:
version: 1.0.0
protocol: legacy_http
enabled: true
idempotent: true
cache_ttl_seconds: 300
timeout_seconds: 30
retry: {enabled: true, max_attempts: 2, backoff_ms: 250}
allowed_agents: []
allowed_channels: []
required_business_keys: []
servers:
telecom:
enabled: true
discover: false
protocol: legacy_http
transport: http
# Local run: uvicorn mcp.servers.telecom_mcp_server.main:app --port 8100
@@ -12,6 +39,7 @@ servers:
retail:
enabled: true
discover: false
protocol: legacy_http
transport: http
# Local run: uvicorn mcp.servers.retail_mcp_server.main:app --port 8200
@@ -170,3 +198,19 @@ auth:
static_tokens:
runtime-local-token:
agents: []
# Example of a dynamically discovered MCP server.
# Uncomment and adjust the URL to plug a new server that exposes a catalog/manifest.
# servers:
# nf_items:
# enabled: true
# discover: true
# protocol: legacy_http
# transport: http
# url: http://localhost:8400/mcp
# # Optional. If omitted, the gateway tries the discovery.default_catalog_endpoints.
# # manifest_url: http://localhost:8400/.well-known/mcp-server.json
# # catalog_endpoint: /tools
# # invoke_endpoint: /tools/call
# timeout_seconds: 30

View File

@@ -0,0 +1,156 @@
# MCP Gateway — Server Discovery and Catalog Sync
## Goal
This evolution allows the MCP Gateway to discover tools from registered MCP Servers by reading a manifest or catalog endpoint.
The framework still points to a single MCP Gateway:
```env
MCP_GATEWAY_ENABLED=true
MCP_GATEWAY_URL=http://localhost:8300
MCP_GATEWAY_TIMEOUT_SECONDS=60
```
The MCP Gateway can point to many MCP Servers:
```text
Agent Framework
-> MCP Gateway
-> telecom_mcp_server
-> retail_mcp_server
-> nf_items_mcp_server
-> any other MCP Server
```
## What is automatic
After a server is registered in `apps/mcp_gateway/config/mcp_gateway.yaml` with `discover: true`, the gateway can:
- call its manifest/catalog endpoint;
- normalize the returned tool list;
- publish the tools in `GET /v1/tools`;
- execute the discovered tool through `POST /v1/tools/{tool_name}/invoke`.
## What is still explicit
The gateway does not scan the network or GitHub by itself. You still register the MCP Server endpoint in YAML.
Example:
```yaml
servers:
nf_items:
enabled: true
discover: true
protocol: legacy_http
transport: http
url: http://localhost:8400/mcp
catalog_endpoint: /tools
invoke_endpoint: /tools/call
timeout_seconds: 30
```
If `catalog_endpoint` is omitted, the gateway tries:
```text
/.well-known/mcp-server.json
/manifest
/mcp/tools
/tools/list
/tools
/v1/tools
```
## Expected manifest/catalog formats
The gateway accepts common shapes:
```json
{
"server_id": "nf_items",
"tools": [
{
"name": "buscar_notas_por_criterios",
"description": "Search invoice items by criteria.",
"input_schema": {
"cliente": "string",
"estado": "string",
"preco": "number",
"ean": "string",
"margem": "number"
}
}
]
}
```
It also accepts:
```json
{"tools": [...]}
```
```json
{"data": {"tools": [...]}}
```
```json
{"capabilities": {"tools": [...]}}
```
## New endpoints
### List discovery servers
```bash
curl http://localhost:8300/v1/discovery/servers | jq
```
### Force catalog sync
```bash
curl -X POST http://localhost:8300/v1/discovery/sync | jq
```
### List merged static + discovered tools
```bash
curl http://localhost:8300/v1/tools | jq
```
## Precedence rule
Static tools configured under `tools:` override discovered tools with the same name. This allows operations teams to override timeout, cache, allowed agents, required business keys, and endpoint behavior safely.
## Plugging a new MCP Server
1. Start the MCP Server.
2. Confirm that it exposes a catalog or manifest endpoint.
3. Add it under `servers:` in `mcp_gateway.yaml` with `discover: true`.
4. Restart the MCP Gateway or call `POST /v1/discovery/sync`.
5. Confirm the tool appears in `GET /v1/tools`.
6. Invoke the tool through the gateway.
## Example invocation
```bash
curl -s -X POST http://localhost:8300/v1/tools/buscar_notas_por_criterios/invoke \
-H "Content-Type: application/json" \
-d '{
"tenant_id": "default",
"agent_id": "telecom_contas",
"channel": "web",
"tool_name": "buscar_notas_por_criterios",
"arguments": {
"cliente": "CLIENTE-001",
"estado": "SP",
"preco": 100.0,
"ean": "7890000000000",
"margem": 0.05
},
"business_context": {
"session_key": "session-001"
}
}' | jq
```

View File

@@ -99,6 +99,9 @@ class Settings(BaseSettings):
OCI_EMBEDDING_MODEL: str = 'cohere.embed-multilingual-v3.0'
ENABLE_LANGFUSE: bool = False
LANGFUSE_TRACE_MODE: Literal['verbose','compact'] = 'verbose'
LANGFUSE_ROOT_SPAN_NAME: str = 'agent.gateway_message'
LANGFUSE_LEGACY_IO_FALLBACK: bool = True
LANGFUSE_PUBLIC_KEY: str | None = None
LANGFUSE_SECRET_KEY: str | None = None
LANGFUSE_HOST: str = 'https://cloud.langfuse.com'

View File

@@ -74,20 +74,23 @@ class MockLLMProvider(LLMProvider):
profile_found = kwargs.get("profile_found")
profiles_enabled = kwargs.get("profiles_enabled")
profiles_path = kwargs.get("profiles_path")
last = messages[-1].get("content", "") if messages else ""
answer = f"[mock-llm] Resposta simulada para: {last[:300]}"
usage = {"prompt_tokens": max(1, len(str(messages))//4), "completion_tokens": max(1, len(answer)//4), "total_tokens": max(2, (len(str(messages))+len(answer))//4), "cost_usd": 0.0, "cost_brl": 0.0}
llm_metadata = {"provider": "mock", "profile_name": profile_name, "component": component_name, "model": model, "profile_source": profile_source, "profile_found": profile_found, "profiles_enabled": profiles_enabled, "profiles_path": profiles_path}
async with _maybe_generation(
self.telemetry,
name=generation_name,
model=model,
input=messages,
metadata=llm_metadata,
model_parameters={},
) as generation:
last = messages[-1].get("content", "") if messages else ""
answer = f"[mock-llm] Resposta simulada para: {last[:300]}"
usage = {"prompt_tokens": max(1, len(str(messages))//4), "completion_tokens": max(1, len(answer)//4), "total_tokens": max(2, (len(str(messages))+len(answer))//4), "cost_usd": 0.0, "cost_brl": 0.0}
generation.set_output(answer)
generation.set_usage(usage)
generation.set_metadata(**usage)
if self.usage_repository:
await self.usage_repository.record(UsageRecord.from_usage("mock", model, generation_name, usage, {"provider":"mock", "profile_name": profile_name, "component": component_name, "model": model, "profile_source": profile_source, "profile_found": profile_found, "profiles_enabled": profiles_enabled, "profiles_path": profiles_path}))
if self.telemetry:
await self.telemetry.generation(
name=generation_name,
model=model,
input=messages,
output=answer,
metadata={"provider": "mock", "profile_name": profile_name, "component": component_name, "model": model, "profile_source": profile_source, "profile_found": profile_found, "profiles_enabled": profiles_enabled, "profiles_path": profiles_path, **usage},
usage=usage,
)
await self.usage_repository.record(UsageRecord.from_usage("mock", model, generation_name, usage, llm_metadata))
return answer
@@ -259,6 +262,22 @@ class OCICompatibleOpenAIProvider(LLMProvider):
for optional_key in ("top_p", "frequency_penalty", "presence_penalty"):
if effective.get(optional_key) is not None:
request_kwargs[optional_key] = effective[optional_key]
model_parameters = {
key: value
for key, value in request_kwargs.items()
if key not in {"model", "messages"} and value is not None
}
llm_metadata = {
"provider": provider,
"model": model,
"component": component_name,
"profile_name": resolved_profile_name,
"requested_profile_name": requested_profile_name,
"profile_source": profile_source,
"profile_found": profile_found,
"profiles_enabled": bool(effective.get("profiles_enabled")),
"profiles_path": effective.get("profiles_path"),
}
async with _maybe_span(
self.telemetry,
@@ -275,47 +294,35 @@ class OCICompatibleOpenAIProvider(LLMProvider):
profiles_enabled=bool(effective.get("profiles_enabled")),
):
try:
resp = await client.chat.completions.create(**request_kwargs)
answer = resp.choices[0].message.content or ""
async with _maybe_generation(
self.telemetry,
name=generation_name,
model=model,
input=messages,
metadata=llm_metadata,
model_parameters=model_parameters,
) as generation:
resp = await client.chat.completions.create(**request_kwargs)
answer = resp.choices[0].message.content or ""
usage_metadata = self.token_collector.enrich(model, getattr(resp, "usage", None))
usage_metadata.update({
"profile_name": resolved_profile_name,
"requested_profile_name": requested_profile_name,
"profile_source": profile_source,
"profile_found": profile_found,
"component": component_name,
"model": model,
"provider": provider,
"temperature": temperature,
"max_tokens": max_tokens,
})
llm_metadata = {
"provider": provider,
"model": model,
"component": component_name,
"profile_name": resolved_profile_name,
"requested_profile_name": requested_profile_name,
"profile_source": profile_source,
"profile_found": profile_found,
"profiles_enabled": bool(effective.get("profiles_enabled")),
"profiles_path": effective.get("profiles_path"),
"temperature": temperature,
"max_tokens": max_tokens,
}
usage_metadata = self.token_collector.enrich(model, getattr(resp, "usage", None))
usage_metadata.update({
"profile_name": resolved_profile_name,
"requested_profile_name": requested_profile_name,
"profile_source": profile_source,
"profile_found": profile_found,
"component": component_name,
"model": model,
"provider": provider,
**model_parameters,
})
generation.set_output(answer)
generation.set_usage(usage_metadata)
generation.set_metadata(**usage_metadata)
if self.usage_repository:
await self.usage_repository.record(
UsageRecord.from_usage(provider, model, generation_name, usage_metadata, llm_metadata)
)
if self.telemetry:
await self.telemetry.generation(
name=generation_name,
model=model,
input=messages,
output=answer,
metadata={**llm_metadata, **usage_metadata},
usage=usage_metadata,
)
return answer
except Exception as exc:
@@ -550,6 +557,19 @@ class OCISDKProvider(LLMProvider):
)
service_endpoint = self._resolve_endpoint(self.settings, endpoint)
model_parameters = {
"temperature": temperature,
"max_tokens": max_tokens,
}
llm_metadata = {
"provider": "oci_sdk",
"model": model,
"endpoint_id": endpoint_id,
"service_endpoint": service_endpoint,
"component": component_name,
"profile_name": profile_name,
"auth_mode": getattr(self.settings, "OCI_AUTH_MODE", "config_file"),
}
async with _maybe_span(
self.telemetry,
@@ -575,26 +595,30 @@ class OCISDKProvider(LLMProvider):
max_tokens=max_tokens,
)
response = await asyncio.to_thread(client.chat, details)
answer = self._extract_answer(response)
async with _maybe_generation(
self.telemetry,
name=generation_name,
model=model,
input=messages,
metadata=llm_metadata,
model_parameters=model_parameters,
) as generation:
response = await asyncio.to_thread(client.chat, details)
answer = self._extract_answer(response)
usage_metadata = {
"prompt_tokens": max(1, len(str(messages)) // 4),
"completion_tokens": max(1, len(answer) // 4),
"total_tokens": max(2, (len(str(messages)) + len(answer)) // 4),
"cost_usd": 0.0,
"cost_brl": 0.0,
"estimated_usage": True,
"provider": "oci_sdk",
"model": model,
"endpoint_id": endpoint_id,
"service_endpoint": service_endpoint,
"component": component_name,
"profile_name": profile_name,
"auth_mode": getattr(self.settings, "OCI_AUTH_MODE", "config_file"),
}
llm_metadata = dict(usage_metadata)
usage_metadata = {
"prompt_tokens": max(1, len(str(messages)) // 4),
"completion_tokens": max(1, len(answer) // 4),
"total_tokens": max(2, (len(str(messages)) + len(answer)) // 4),
"cost_usd": 0.0,
"cost_brl": 0.0,
"estimated_usage": True,
**llm_metadata,
**model_parameters,
}
generation.set_output(answer)
generation.set_usage(usage_metadata)
generation.set_metadata(**usage_metadata)
if self.usage_repository:
await self.usage_repository.record(
@@ -607,16 +631,6 @@ class OCISDKProvider(LLMProvider):
)
)
if self.telemetry:
await self.telemetry.generation(
name=generation_name,
model=model,
input=messages,
output=answer,
metadata=llm_metadata,
usage=usage_metadata,
)
return answer
@@ -656,3 +670,36 @@ class _maybe_span:
if self.cm:
return await self.cm.__aexit__(exc_type, exc, tb)
return False
class _NoopGeneration:
def set_output(self, output: Any) -> None:
pass
def set_usage(self, usage: dict[str, Any] | None) -> None:
pass
def set_metadata(self, **metadata: Any) -> None:
pass
def set_model_parameters(self, **model_parameters: Any) -> None:
pass
class _maybe_generation:
def __init__(self, telemetry, **attrs: Any):
self.telemetry = telemetry
self.attrs = attrs
self.cm = None
self.noop = _NoopGeneration()
async def __aenter__(self):
if not self.telemetry or not hasattr(self.telemetry, "generation_span"):
return self.noop
self.cm = self.telemetry.generation_span(**self.attrs)
return await self.cm.__aenter__()
async def __aexit__(self, exc_type, exc, tb):
if self.cm:
return await self.cm.__aexit__(exc_type, exc, tb)
return False

View File

@@ -21,6 +21,7 @@ _workflow_id: ContextVar[str | None] = ContextVar("workflow_id", default=None)
_message_id: ContextVar[str | None] = ContextVar("message_id", default=None)
_trace_id: ContextVar[str | None] = ContextVar("trace_id", default=None)
_current_observation_id: ContextVar[str | None] = ContextVar("current_observation_id", default=None)
_current_span_events: ContextVar[list[dict[str, Any]] | None] = ContextVar("current_span_events", default=None)
@dataclass(slots=True)
class ObservabilityContext:
@@ -66,6 +67,31 @@ def reset_current_observation_id(token) -> None:
_current_observation_id.set(None)
def get_current_span_events() -> list[dict[str, Any]] | None:
"""Return the mutable aggregate-event bucket for the active macro span."""
return _current_span_events.get()
def set_current_span_events(events: list[dict[str, Any]] | None):
"""Set aggregate-event bucket and return ContextVar token."""
return _current_span_events.set(events)
def reset_current_span_events(token) -> None:
"""Restore previous aggregate-event bucket."""
try:
_current_span_events.reset(token)
except Exception:
_current_span_events.set(None)
def record_current_span_event(event: dict[str, Any]) -> None:
"""Append an event summary to the active macro span, if one exists."""
events = _current_span_events.get()
if events is not None:
events.append(event)
def set_observability_context(**kwargs: Any) -> ObservabilityContext:
if not kwargs.get("request_id") and not _request_id.get():
kwargs["request_id"] = str(uuid4())
@@ -82,7 +108,7 @@ def set_observability_context(**kwargs: Any) -> ObservabilityContext:
def clear_observability_context() -> None:
for var in (_request_id, _session_id, _user_id, _tenant_id, _agent_id, _channel, _ura_call_id, _workflow_id, _message_id, _trace_id, _current_observation_id):
for var in (_request_id, _session_id, _user_id, _tenant_id, _agent_id, _channel, _ura_call_id, _workflow_id, _message_id, _trace_id, _current_observation_id, _current_span_events):
var.set(None)

View File

@@ -3,6 +3,36 @@ import time
from contextlib import asynccontextmanager
from typing import Any
_LANGGRAPH_STEP_ORDER = {
"__start__": 0,
"input_guardrails": 1,
"routing_decision": 2,
"billing_agent": 3,
"product_agent": 3,
"orders_agent": 3,
"support_agent": 3,
"handoff": 3,
"supervisor_agent": 3,
"output_supervisor": 4,
"output_guardrails": 5,
"judge": 6,
"supervisor_review": 7,
"persist": 8,
"__end__": 9,
}
def _langgraph_step(name: str, state: dict[str, Any]) -> int:
explicit_steps = state.get("langgraph_steps")
if isinstance(explicit_steps, dict) and name in explicit_steps:
try:
return int(explicit_steps[name])
except (TypeError, ValueError):
pass
return _LANGGRAPH_STEP_ORDER.get(name, 50)
class LangGraphDeepTelemetry:
"""Eventos profundos do LangGraph no padrão FIRST.
@@ -16,7 +46,16 @@ class LangGraphDeepTelemetry:
async def node(self, name: str, state: dict[str, Any] | None = None):
state=state or {}
session_id=state.get('conversation_key') or state.get('session_id')
payload={'node': name, 'session_id': session_id, 'agent_id': state.get('agent_id'), 'tenant_id': state.get('tenant_id'), 'input_size': len(str(state.get('user_text') or state.get('sanitized_input') or ''))}
payload={
'node': name,
'langgraph_node': name,
'langgraph_step': _langgraph_step(name, state),
'framework': 'langgraph',
'session_id': session_id,
'agent_id': state.get('agent_id'),
'tenant_id': state.get('tenant_id'),
'input_size': len(str(state.get('user_text') or state.get('sanitized_input') or '')),
}
start=time.time()
await self.telemetry.event('langgraph.node.started', payload, kind='langgraph')
async with self.telemetry.span(f'langgraph.node.{name}', **payload):

View File

@@ -15,15 +15,20 @@ import logging
import re
import time
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Any
from uuid import uuid4
from .context import (
context_metadata,
get_current_observation_id,
get_current_span_events,
get_observability_context,
record_current_span_event,
reset_current_observation_id,
reset_current_span_events,
set_current_observation_id,
set_current_span_events,
set_observability_context,
)
from .event_bus import TelemetryEventBus
@@ -32,6 +37,24 @@ from .otel import OpenTelemetryProvider
logger = logging.getLogger("agent_framework.telemetry")
_LANGFUSE_OBSERVATION_TYPES = {"span", "generation", "agent", "tool", "chain", "retriever", "embedding", "evaluator", "guardrail"}
_LANGFUSE_START_OBSERVATION_KWARGS = {
"trace_context",
"name",
"as_type",
"input",
"output",
"metadata",
"version",
"level",
"status_message",
"completion_start_time",
"model",
"model_parameters",
"usage_details",
"cost_details",
"prompt",
"end_on_exit",
}
def _langfuse_type(kind: str | None) -> str:
# Langfuse SDKs do not accept arbitrary event types such as "event"; FIRST pattern
@@ -42,6 +65,19 @@ def _langfuse_type(kind: str | None) -> str:
_LANGFUSE_TRACE_ID_RE = re.compile(r"^[0-9a-f]{32}$")
_COMPACT_SUPPRESSED_SPAN_PREFIXES = (
"llm.chat_completion",
"workflow.agent.",
"workflow.handoff",
"workflow.input_guardrails",
"workflow.judge",
"workflow.output_guardrails",
"workflow.output_supervisor",
"workflow.persist",
"workflow.routing_decision",
"workflow.supervisor_review",
)
_COMPACT_VISIBLE_EVENT_PREFIXES = ("AGA.", "NOC.")
def _raw_correlation_id(attrs: dict[str, Any] | None = None) -> str | None:
@@ -100,6 +136,7 @@ def _inject_langfuse_trace_context(kwargs: dict[str, Any], attrs: dict[str, Any]
which grouped everything in one trace while flattening the tree.
"""
attrs = attrs or kwargs.get("metadata") or {}
ignore_current_parent = bool(attrs.get("_ignore_current_parent") or kwargs.get("_ignore_current_parent"))
raw_id = _raw_correlation_id(attrs)
trace_id = _langfuse_trace_id(raw_id)
parent_id = (
@@ -107,7 +144,7 @@ def _inject_langfuse_trace_context(kwargs: dict[str, Any], attrs: dict[str, Any]
or attrs.get("parent_span_id")
or kwargs.get("parent_observation_id")
or kwargs.get("parent_span_id")
or get_current_observation_id()
or (None if ignore_current_parent else get_current_observation_id())
)
if trace_id:
trace_context = dict(kwargs.get("trace_context") or {})
@@ -121,6 +158,8 @@ def _inject_langfuse_trace_context(kwargs: dict[str, Any], attrs: dict[str, Any]
metadata.setdefault("langfuse_trace_id", trace_id)
if parent_id:
metadata.setdefault("parent_observation_id", str(parent_id))
metadata.pop("_ignore_current_parent", None)
kwargs.pop("_ignore_current_parent", None)
return kwargs
@@ -150,6 +189,128 @@ def _extract_observation_id(observation: Any) -> str | None:
pass
return None
def _is_compact_visible_event(name: str) -> bool:
return str(name or "").startswith(_COMPACT_VISIBLE_EVENT_PREFIXES)
class _SpanHandle:
"""Mutable handle yielded by Telemetry.span for setting final output."""
def __init__(self, observation: Any | None = None) -> None:
self.observation = observation
self.output: Any = None
self.has_output = False
self.metadata: dict[str, Any] = {}
def set_observation(self, observation: Any | None) -> None:
self.observation = observation
def set_output(self, output: Any) -> None:
self.output = output
self.has_output = True
def set_metadata(self, **metadata: Any) -> None:
self.metadata.update({k: v for k, v in metadata.items() if v is not None})
def __getattr__(self, name: str) -> Any:
if self.observation is None:
raise AttributeError(name)
return getattr(self.observation, name)
class _GenerationHandle:
"""Mutable handle yielded by Telemetry.generation_span."""
def __init__(self, observation: Any | None = None) -> None:
self.observation = observation
self.output: Any = None
self.has_output = False
self.metadata: dict[str, Any] = {}
self.usage: dict[str, Any] | None = None
self.model_parameters: dict[str, Any] = {}
def set_observation(self, observation: Any | None) -> None:
self.observation = observation
def set_output(self, output: Any) -> None:
self.output = output
self.has_output = True
def set_usage(self, usage: dict[str, Any] | None) -> None:
self.usage = dict(usage or {})
def set_metadata(self, **metadata: Any) -> None:
self.metadata.update({k: v for k, v in metadata.items() if v is not None})
def set_model_parameters(self, **model_parameters: Any) -> None:
self.model_parameters.update({k: v for k, v in model_parameters.items() if v is not None})
def __getattr__(self, name: str) -> Any:
if self.observation is None:
raise AttributeError(name)
return getattr(self.observation, name)
def _usage_details_from_usage(usage: dict[str, Any] | None) -> dict[str, int] | None:
if not isinstance(usage, dict):
return None
def int_value(*keys: str) -> int | None:
for key in keys:
value = usage.get(key)
if value is None:
continue
try:
return int(value)
except (TypeError, ValueError):
continue
return None
input_tokens = int_value("input", "input_tokens", "prompt_tokens")
output_tokens = int_value("output", "output_tokens", "completion_tokens")
total_tokens = int_value("total", "total_tokens")
# Langfuse self-hosted versions may sum all custom usage keys into totalUsage.
# Send split fields only when available; send total only when there is no split.
details: dict[str, int] = {}
if input_tokens is not None:
details["input"] = input_tokens
if output_tokens is not None:
details["output"] = output_tokens
if not details and total_tokens is not None:
details["total"] = total_tokens
return details or None
def _cost_details_from_usage(usage: dict[str, Any] | None) -> dict[str, float] | None:
if not isinstance(usage, dict):
return None
details: dict[str, float] = {}
if usage.get("cost_usd") is not None:
try:
details["total"] = float(usage["cost_usd"])
except (TypeError, ValueError):
pass
if usage.get("cost_brl") is not None:
try:
details["total_brl"] = float(usage["cost_brl"])
except (TypeError, ValueError):
pass
return details or None
def _clean_mapping(value: dict[str, Any] | None) -> dict[str, Any] | None:
if not isinstance(value, dict):
return None
clean = {k: v for k, v in value.items() if v is not None}
return clean or None
def _utc_iso_ms() -> str:
return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
class Telemetry:
def __init__(self, settings):
self.settings = settings
@@ -188,6 +349,15 @@ class Telemetry:
def is_enabled(self) -> bool:
return bool(self.enabled and self.langfuse)
def is_compact_mode(self) -> bool:
mode = getattr(self.settings, "LANGFUSE_TRACE_MODE", "verbose") or "verbose"
return str(mode).lower() == "compact"
def _should_emit_langfuse_span(self, name: str) -> bool:
if not self.is_compact_mode():
return True
return not str(name).startswith(_COMPACT_SUPPRESSED_SPAN_PREFIXES)
def bind_context(self, **kwargs: Any):
return set_observability_context(**kwargs)
@@ -199,6 +369,10 @@ class Telemetry:
"""Cria span correlacionado em logs, Langfuse e OpenTelemetry."""
start = time.time()
attrs = context_metadata(attrs)
attrs.setdefault("_span_name", name)
is_root_span = bool(attrs.get("_root_span")) or name == "agent.gateway_message"
if self.is_compact_mode() and is_root_span and not attrs.get("parent_observation_id"):
attrs["_ignore_current_parent"] = True
if not attrs.get("request_id"):
attrs["request_id"] = str(uuid4())
if not attrs.get("trace_id"):
@@ -206,45 +380,90 @@ class Telemetry:
set_observability_context(request_id=attrs.get("request_id"), trace_id=attrs.get("trace_id"))
observation_cm = None
observation = None
handle = _SpanHandle()
observation_token = None
parent_observation_id = attrs.get("parent_observation_id") or get_current_observation_id()
propagation_cm = None
legacy_io_update: dict[str, Any] | None = None
ignore_current_parent = bool(attrs.get("_ignore_current_parent"))
parent_observation_id = attrs.get("parent_observation_id")
if not parent_observation_id and not ignore_current_parent:
parent_observation_id = get_current_observation_id()
if parent_observation_id:
attrs.setdefault("parent_observation_id", str(parent_observation_id))
logger.info("span.start %s %s", name, _safe(attrs))
otel_cm = self.otel.span(name, attrs)
otel_span = otel_cm.__enter__()
if self.is_enabled():
emit_langfuse_span = self.is_enabled() and self._should_emit_langfuse_span(name)
span_events: list[dict[str, Any]] | None = [] if emit_langfuse_span and self.is_compact_mode() else None
span_events_token = set_current_span_events(span_events) if span_events is not None else None
observation_metadata = {k: v for k, v in attrs.items() if k != "input" and not str(k).startswith("_")}
if emit_langfuse_span:
observation_cm = self._start_observation(
name=name,
as_type="span",
input=attrs.get("input"),
metadata={k: v for k, v in attrs.items() if k != "input"},
metadata=observation_metadata,
_ignore_current_parent=attrs.get("_ignore_current_parent"),
)
try:
if observation_cm is not None:
observation = observation_cm.__enter__()
handle.set_observation(observation)
observation_id = _extract_observation_id(observation)
if observation_id:
observation_token = set_current_observation_id(observation_id)
attrs.setdefault("observation_id", observation_id)
self._update_trace_from_attrs(observation, attrs)
if is_root_span:
self._update_trace_from_attrs(observation, attrs)
self._set_trace_io(observation, input=attrs.get("input"))
propagation_cm = self._start_trace_attribute_propagation(name, attrs)
if propagation_cm is not None:
propagation_cm.__enter__()
# Publish span.started only after the Langfuse observation is current,
# so secondary analytics/exporters can attach it as a child instead
# of creating a sibling/root entry.
await self.event_bus.publish(f"{name}.started", attrs, kind="span")
yield observation
yield handle
duration_ms = int((time.time() - start) * 1000)
out = {"status": "ok", "duration_ms": duration_ms}
self._update_observation(observation, output=out, metadata={"duration_ms": duration_ms})
status = {"status": "ok", "duration_ms": duration_ms}
out = handle.output if handle.has_output else status
metadata = {**observation_metadata, **status, **handle.metadata}
if span_events is not None:
metadata["aggregated_event_count"] = len(span_events)
metadata["aggregated_events"] = span_events
self._update_observation(observation, input=attrs.get("input"), output=out, metadata=metadata)
if is_root_span:
self._set_trace_io(observation, input=attrs.get("input"), output=out)
legacy_io_update = {
"input": attrs.get("input"),
"output": out,
"metadata": metadata,
}
if otel_span is not None:
otel_span.set_attribute("duration_ms", duration_ms)
await self.event_bus.publish(f"{name}.completed", {**attrs, **out}, kind="span")
completed_payload = {**attrs, **status}
if handle.has_output:
completed_payload["output"] = out
await self.event_bus.publish(f"{name}.completed", completed_payload, kind="span")
logger.info("span.end %s duration_ms=%s", name, duration_ms)
except Exception as exc:
duration_ms = int((time.time() - start) * 1000)
out = {"status": "error", "error": str(exc), "duration_ms": duration_ms}
self._update_observation(observation, level="ERROR", status_message=str(exc), output=out, metadata={"duration_ms": duration_ms})
metadata = {**observation_metadata, "duration_ms": duration_ms}
if span_events is not None:
metadata["aggregated_event_count"] = len(span_events)
metadata["aggregated_events"] = span_events
self._update_observation(observation, level="ERROR", status_message=str(exc), input=attrs.get("input"), output=out, metadata=metadata)
if is_root_span:
self._set_trace_io(observation, input=attrs.get("input"), output=out)
legacy_io_update = {
"input": attrs.get("input"),
"output": out,
"metadata": metadata,
"level": "ERROR",
"status_message": str(exc),
}
if otel_span is not None:
try:
otel_span.record_exception(exc)
@@ -255,11 +474,23 @@ class Telemetry:
logger.exception("span.error %s %s", name, exc)
raise
finally:
if propagation_cm is not None:
try: propagation_cm.__exit__(None, None, None)
except Exception: logger.debug("Falha ao encerrar propagação Langfuse", exc_info=True)
if observation_cm is not None:
try: observation_cm.__exit__(None, None, None)
except Exception: logger.exception("Falha ao finalizar span Langfuse %s", name)
if legacy_io_update is not None:
self._legacy_observation_update(
observation,
observation_type="span",
name=name,
**legacy_io_update,
)
if observation_token is not None:
reset_current_observation_id(observation_token)
if span_events_token is not None:
reset_current_span_events(span_events_token)
try: otel_cm.__exit__(None, None, None)
except Exception: logger.debug("Falha ao fechar span OTEL", exc_info=True)
@@ -267,6 +498,24 @@ class Telemetry:
payload = context_metadata(payload or {})
logger.info("event %s %s", name, _safe(payload))
await self.event_bus.publish(name, payload, kind=kind)
if self.is_compact_mode():
if get_current_span_events() is not None:
record_current_span_event({
"name": name,
"kind": kind,
"payload": payload,
})
if not _is_compact_visible_event(name) or not self.is_enabled():
return
try:
metadata = {**payload, "event_kind": kind}
cm = self._start_observation(name=name, as_type="span", input=payload, metadata=metadata)
if cm is not None:
with cm as obs:
self._update_observation(obs, input=payload, output={"status": "ok"}, metadata=metadata)
except Exception:
logger.exception("Falha ao enviar event compacto via observation")
return
if not self.is_enabled():
return
# IMPORTANT: do not call ``langfuse.event(...)`` directly here. In SDK
@@ -274,54 +523,179 @@ class Telemetry:
# a new trace row for every telemetry event. We create a correlated
# observation instead, using request_id/trace_id as the stable trace id.
try:
cm = self._start_observation(name=name, as_type=_langfuse_type(kind), metadata={**payload, "event_kind": kind})
metadata = {**payload, "event_kind": kind}
if self.is_compact_mode():
metadata["_ignore_current_parent"] = True
cm = self._start_observation(name=name, as_type=_langfuse_type(kind), metadata=metadata)
if cm is not None:
with cm: pass
except Exception:
logger.exception("Falha ao enviar event via observation")
async def generation(self, name: str, model: str, input: list | dict | str, output: str,
metadata: dict[str, Any] | None = None, usage: dict[str, Any] | None = None):
@asynccontextmanager
async def generation_span(
self,
name: str,
model: str,
input: list | dict | str,
*,
metadata: dict[str, Any] | None = None,
usage: dict[str, Any] | None = None,
model_parameters: dict[str, Any] | None = None,
):
metadata = context_metadata(metadata or {})
# Keep the actual LLM model visible both in Langfuse's generation.model field
# and in metadata for filtering/debugging across SDK versions.
metadata.setdefault("model", model)
metadata.setdefault("llm_model", model)
metadata.setdefault("component", metadata.get("profile_name") or name)
if usage:
metadata["usage"] = usage
logger.info("generation %s model=%s component=%s profile=%s metadata=%s", name, model, metadata.get("component"), metadata.get("profile_name"), _safe(metadata))
await self.event_bus.publish(name, {"model": model, "llm_model": model, "output_chars": len(output or ""), **metadata}, kind="generation")
if not self.is_enabled():
return
clean_model_parameters = _clean_mapping(model_parameters)
if clean_model_parameters:
metadata.setdefault("model_parameters", clean_model_parameters)
handle = _GenerationHandle()
observation_cm = None
observation = None
observation_token = None
legacy_io_update: dict[str, Any] | None = None
logger.info("generation.start %s model=%s component=%s profile=%s metadata=%s", name, model, metadata.get("component"), metadata.get("profile_name"), _safe(metadata))
try:
kwargs = dict(name=name, as_type="generation", input=input, output=output, model=model, metadata=metadata)
if usage:
kwargs["usage"] = usage
kwargs["usage_details"] = {k: usage.get(k) for k in ("prompt_tokens", "completion_tokens", "total_tokens", "cached_tokens", "reasoning_tokens") if k in usage}
# Prefer current/correlated generation APIs. Avoid raw
# ``langfuse.generation(...)`` first because it can create a separate
# trace row per LLM call when no current observation exists.
if hasattr(self.langfuse, "start_as_current_generation"):
clean = {k: v for k, v in kwargs.items() if k != "as_type" and v is not None}
clean = _inject_langfuse_trace_context(clean, metadata)
if self.is_enabled():
try:
with self.langfuse.start_as_current_generation(**clean) as obs:
self._update_observation(obs, output=output, model=model, metadata=metadata)
return
except TypeError:
clean.pop("trace_context", None)
with self.langfuse.start_as_current_generation(**clean) as obs:
self._update_observation(obs, output=output, model=model, metadata=metadata)
return
observation_cm = self._start_observation(
name=name,
as_type="generation",
input=input,
model=model,
model_parameters=clean_model_parameters,
usage_details=_usage_details_from_usage(usage),
cost_details=_cost_details_from_usage(usage),
metadata=metadata,
)
if observation_cm is not None:
observation = observation_cm.__enter__()
handle.set_observation(observation)
observation_id = _extract_observation_id(observation)
if observation_id:
observation_token = set_current_observation_id(observation_id)
except Exception:
observation_cm = None
observation = None
logger.exception("Falha ao iniciar generation Langfuse %s", name)
yield handle
final_usage = handle.usage if handle.usage is not None else usage
final_model_parameters = {
**(clean_model_parameters or {}),
**handle.model_parameters,
} or None
final_metadata = {**metadata, **handle.metadata}
if final_usage:
final_metadata["usage"] = final_usage
output = handle.output if handle.has_output else None
usage_details = _usage_details_from_usage(final_usage)
cost_details = _cost_details_from_usage(final_usage)
self._update_observation(
observation,
input=input,
output=output,
model=model,
metadata=final_metadata,
model_parameters=final_model_parameters,
usage_details=usage_details,
cost_details=cost_details,
)
legacy_io_update = {
"input": input,
"output": output,
"model": model,
"metadata": final_metadata,
"model_parameters": final_model_parameters,
"usage_details": usage_details,
"cost_details": cost_details,
}
await self.event_bus.publish(
name,
{
"model": model,
"llm_model": model,
"output_chars": len(output or "") if isinstance(output, str) else 0,
**final_metadata,
},
kind="generation",
)
logger.info("generation.end %s model=%s", name, model)
except Exception as exc:
final_usage = handle.usage if handle.usage is not None else usage
final_model_parameters = {
**(clean_model_parameters or {}),
**handle.model_parameters,
} or None
final_metadata = {**metadata, **handle.metadata}
if final_usage:
final_metadata["usage"] = final_usage
usage_details = _usage_details_from_usage(final_usage)
cost_details = _cost_details_from_usage(final_usage)
output = handle.output if handle.has_output else None
self._update_observation(
observation,
level="ERROR",
status_message=str(exc),
input=input,
output=output,
model=model,
metadata=final_metadata,
model_parameters=final_model_parameters,
usage_details=usage_details,
cost_details=cost_details,
)
legacy_io_update = {
"input": input,
"output": output,
"model": model,
"metadata": final_metadata,
"model_parameters": final_model_parameters,
"usage_details": usage_details,
"cost_details": cost_details,
"level": "ERROR",
"status_message": str(exc),
}
await self.event_bus.publish(f"{name}.failed", {"model": model, "llm_model": model, "error": str(exc), **final_metadata}, kind="generation")
logger.exception("generation.error %s model=%s exc=%s", name, model, exc)
raise
finally:
if observation_cm is not None:
try: observation_cm.__exit__(None, None, None)
except Exception: logger.exception("Falha ao finalizar generation Langfuse %s", name)
if legacy_io_update is not None:
self._legacy_observation_update(
observation,
observation_type="generation",
name=name,
**legacy_io_update,
)
if observation_token is not None:
reset_current_observation_id(observation_token)
cm = self._start_observation(**kwargs)
if cm is not None:
with cm as obs:
self._update_observation(obs, output=output, model=model, metadata=metadata)
except Exception:
logger.exception("Falha ao registrar generation no Langfuse")
async def generation(
self,
name: str,
model: str,
input: list | dict | str,
output: str,
metadata: dict[str, Any] | None = None,
usage: dict[str, Any] | None = None,
model_parameters: dict[str, Any] | None = None,
):
async with self.generation_span(
name=name,
model=model,
input=input,
metadata=metadata,
usage=usage,
model_parameters=model_parameters,
) as generation:
generation.set_output(output)
if usage:
generation.set_usage(usage)
async def rag_event(self, name: str, query: str, results_count: int, metadata: dict[str, Any] | None = None):
await self.event(f"rag.{name}", {"query": query, "results_count": results_count, **(metadata or {})}, kind="rag")
@@ -364,10 +738,16 @@ class Telemetry:
def _start_observation(self, **kwargs):
if not self.is_enabled(): return None
if hasattr(self.langfuse, "start_as_current_observation"):
clean = {k: v for k, v in kwargs.items() if v is not None}
clean = {k: v for k, v in kwargs.items() if v is not None and k in _LANGFUSE_START_OBSERVATION_KWARGS}
if "as_type" in clean:
clean["as_type"] = _langfuse_type(clean.get("as_type"))
clean = _inject_langfuse_trace_context(clean, clean.get("metadata") or {})
if self.is_compact_mode():
clean.pop("_ignore_current_parent", None)
else:
clean = _inject_langfuse_trace_context(clean, clean.get("metadata") or {})
metadata = clean.get("metadata")
if isinstance(metadata, dict):
clean["metadata"] = {k: v for k, v in metadata.items() if not str(k).startswith("_")}
try:
return self.langfuse.start_as_current_observation(**clean)
except (TypeError, ValueError):
@@ -413,9 +793,66 @@ class Telemetry:
if hasattr(observation, "update"): observation.update(**clean)
except Exception: logger.debug("Observation update não suportado", exc_info=True)
def _legacy_observation_update(self, observation, *, observation_type: str, name: str, **kwargs):
"""Compatibility fallback for Langfuse servers that drop OTEL observation I/O."""
if not self.is_enabled() or not bool(getattr(self.settings, "LANGFUSE_LEGACY_IO_FALLBACK", True)):
return
if observation is None:
return
obs_id = _extract_observation_id(observation)
trace_id = getattr(observation, "trace_id", None)
if not obs_id or not trace_id:
return
api = getattr(self.langfuse, "api", None)
ingestion = getattr(api, "ingestion", None)
if ingestion is None or not hasattr(ingestion, "batch"):
return
clean = {k: v for k, v in kwargs.items() if v is not None}
if not any(k in clean for k in ("input", "output", "metadata")):
return
try:
if hasattr(self.langfuse, "flush"):
self.langfuse.flush()
if observation_type == "generation":
from langfuse.api.ingestion.types import (
IngestionEvent_GenerationUpdate,
UpdateGenerationBody,
)
body = UpdateGenerationBody(id=str(obs_id), trace_id=str(trace_id), name=name, **clean)
event = IngestionEvent_GenerationUpdate(
id=str(uuid4()),
timestamp=_utc_iso_ms(),
body=body,
metadata={"source": "agent_framework", "fallback": "legacy_observation_io"},
)
else:
from langfuse.api.ingestion.types import IngestionEvent_SpanUpdate, UpdateSpanBody
body = UpdateSpanBody(id=str(obs_id), trace_id=str(trace_id), name=name, **clean)
event = IngestionEvent_SpanUpdate(
id=str(uuid4()),
timestamp=_utc_iso_ms(),
body=body,
metadata={"source": "agent_framework", "fallback": "legacy_observation_io"},
)
response = ingestion.batch(
batch=[event],
metadata={"source": "agent_framework", "fallback": "legacy_observation_io"},
)
if getattr(response, "errors", None):
logger.debug("Langfuse legacy I/O fallback retornou erros: %s", response.errors)
except Exception:
logger.debug("Falha no fallback legado de input/output Langfuse", exc_info=True)
def _update_trace_from_attrs(self, observation, attrs: dict[str, Any]):
if observation is None: return
trace_attrs = {}
if attrs.get("_span_name"):
trace_attrs["name"] = attrs["_span_name"]
for key in ("session_id", "user_id"):
if attrs.get(key): trace_attrs[key] = attrs[key]
if attrs.get("input"): trace_attrs["input"] = attrs["input"]
@@ -427,6 +864,43 @@ class Telemetry:
if hasattr(observation, "update_trace"): observation.update_trace(**trace_attrs)
except Exception: logger.debug("Trace update não suportado", exc_info=True)
def _set_trace_io(self, observation, *, input: Any | None = None, output: Any | None = None):
if observation is None: return
try:
if hasattr(observation, "set_trace_io"):
observation.set_trace_io(input=input, output=output)
return
if hasattr(observation, "update_trace"):
payload = {}
if input is not None:
payload["input"] = input
if output is not None:
payload["output"] = output
if payload:
observation.update_trace(**payload)
except Exception: logger.debug("Trace input/output update não suportado", exc_info=True)
def _start_trace_attribute_propagation(self, name: str, attrs: dict[str, Any]):
if not self.is_enabled() or not hasattr(self.langfuse, "propagate_attributes"):
return None
metadata = {
k: attrs.get(k)
for k in ("request_id", "trace_id", "agent_id", "tenant_id", "channel", "message_id", "ura_call_id", "workflow_id")
if attrs.get(k)
}
tags = attrs.get("tags") if isinstance(attrs.get("tags"), list) else None
try:
return self.langfuse.propagate_attributes(
user_id=str(attrs["user_id"]) if attrs.get("user_id") is not None else None,
session_id=str(attrs["session_id"]) if attrs.get("session_id") is not None else None,
metadata=metadata or None,
tags=[str(tag) for tag in tags] if tags else None,
trace_name=name,
)
except Exception:
logger.debug("Trace attribute propagation não suportada", exc_info=True)
return None
class _LegacyObservationContext:
def __init__(self, observation): self.observation = observation
def __enter__(self): return self.observation

View File

@@ -31,6 +31,19 @@ TOOLS = {
async def health():
return {"status": "ok", "server": "retail_mcp_server"}
@app.get("/.well-known/mcp-server.json")
async def well_known_manifest():
return {"server_id": "retail", "protocol": "legacy_http", "invoke_endpoint": "/tools/call", "tools": [{"name": name, **cfg} for name, cfg in TOOLS.items()]}
@app.get("/manifest")
async def manifest():
return await well_known_manifest()
@app.get("/mcp/tools")
async def tools_catalog():
return {"tools": [{"name": name, **cfg} for name, cfg in TOOLS.items()]}
@app.get("/mcp/tools/list")
async def list_tools():
return {"tools": [{"name": name, **cfg} for name, cfg in TOOLS.items()]}

View File

@@ -31,6 +31,19 @@ TOOLS = {
async def health():
return {"status": "ok", "server": "telecom_mcp_server"}
@app.get("/.well-known/mcp-server.json")
async def well_known_manifest():
return {"server_id": "telecom", "protocol": "legacy_http", "invoke_endpoint": "/tools/call", "tools": [{"name": name, **cfg} for name, cfg in TOOLS.items()]}
@app.get("/manifest")
async def manifest():
return await well_known_manifest()
@app.get("/mcp/tools")
async def tools_catalog():
return {"tools": [{"name": name, **cfg} for name, cfg in TOOLS.items()]}
@app.get("/mcp/tools/list")
async def list_tools():
return {"tools": [{"name": name, **cfg} for name, cfg in TOOLS.items()]}

View File

@@ -1,218 +0,0 @@
# SPEC-003 — AI Gateway
## Escopo
O AI Gateway executa chamadas de inferência e embedding por contrato padronizado. Ele centraliza provider, modelo, profile, política, fallback, uso, custo, latência e telemetria.
## Endpoints
| Método | Endpoint | Descrição |
|---|---|---|
| `POST` | `/v1/chat/completions` | Geração de texto/chat. |
| `POST` | `/v1/embeddings` | Geração de embeddings. |
| `GET` | `/v1/models` | Lista modelos disponíveis. |
| `GET` | `/v1/profiles` | Lista profiles configurados. |
| `GET` | `/health` | Health check. |
| `GET` | `/ready` | Readiness check. |
## LLMRequest
```json
{
"tenant_id": "default",
"agent_id": "telecom_contas",
"profile": "judge",
"component": "judge",
"operation": "chat_completion",
"messages": [
{"role": "system", "content": "Você é um avaliador."},
{"role": "user", "content": "Avalie a resposta."}
],
"parameters": {
"temperature": 0,
"max_tokens": 800
},
"metadata": {
"request_id": "req-001",
"trace_id": "trace-001",
"session_id": "default:telecom_contas:session-001"
}
}
```
## LLMResponse
```json
{
"provider": "oci_openai",
"model": "openai.gpt-4.1",
"profile": "judge",
"content": "Resultado da geração.",
"usage": {
"input_tokens": 1200,
"output_tokens": 300,
"total_tokens": 1500
},
"latency_ms": 820,
"finish_reason": "stop",
"metadata": {
"fallback_used": false
}
}
```
## Profiles
```yaml
profiles:
default:
provider: oci_openai
model: openai.gpt-4.1
temperature: 0.2
max_tokens: 2048
supervisor:
provider: oci_openai
model: openai.gpt-4.1
temperature: 0
max_tokens: 700
router:
provider: oci_openai
model: openai.gpt-4.1
temperature: 0
max_tokens: 500
guardrail:
provider: oci_openai
model: openai.gpt-4.1
temperature: 0
max_tokens: 600
judge:
provider: oci_openai
model: openai.gpt-4.1
temperature: 0
max_tokens: 800
rag_generation:
provider: oci_openai
model: openai.gpt-4.1
temperature: 0.1
max_tokens: 1800
```
## Providers
| Provider | Autenticação | Uso |
|---|---|---|
| `mock` | Nenhuma | Testes locais. |
| `oci_openai` | API Key / endpoint compatible | OCI GenAI OpenAI-Compatible. |
| `oci_sdk` | OCI Auth Mode | SDK nativo OCI. |
| `openai_compatible` | API Key | Endpoint compatível OpenAI. |
## OCI Authentication
| Variável | Valores |
|---|---|
| `LLM_PROVIDER` | `mock`, `oci_openai`, `oci_sdk`, `openai_compatible` |
| `OCI_AUTH_MODE` | `config_file`, `instance_principal`, `resource_principal`, `workload_identity` |
| `OCI_GENAI_API_KEY` | API key para provider compatível |
## Fallback
```yaml
fallback:
enabled: true
policies:
default:
chain:
- profile: default
- profile: default_low_cost
judge:
enabled: false
```
## Rate Limit
```yaml
rate_limits:
default:
requests_per_minute: 600
tokens_per_minute: 1000000
agents:
telecom_contas:
requests_per_minute: 120
```
## Eventos
| Evento | Descrição |
|---|---|
| `ai.request.received` | Requisição recebida. |
| `ai.profile.resolved` | Profile resolvido. |
| `ai.provider.selected` | Provider selecionado. |
| `ai.completion.started` | Chamada iniciada. |
| `ai.completion.completed` | Chamada concluída. |
| `ai.fallback.used` | Fallback utilizado. |
| `ai.request.failed` | Falha de inferência. |
## Métricas
| Métrica | Dimensões |
|---|---|
| `ai_requests_total` | provider, model, profile, tenant, agent, status |
| `ai_latency_ms` | provider, model, profile |
| `ai_tokens_total` | provider, model, input/output |
| `ai_cost_estimated` | provider, model, tenant, agent |
| `ai_fallback_total` | source_profile, fallback_profile |
## Segurança
- API keys não são gravadas em logs.
- Prompts podem ser mascarados conforme política.
- Providers são autorizados por tenant.
- Workload Identity é usada em Kubernetes quando configurada.
- Modelos bloqueados por política retornam erro controlado.
## Requisitos Não Funcionais
| Categoria | Requisito |
|---|---|
| Disponibilidade | Componentes deployáveis expõem `/health` e `/ready`. |
| Escalabilidade | Apps stateless escalam horizontalmente. Estado conversacional fica em repositórios externos. |
| Segurança | Segredos são fornecidos por secret store ou Kubernetes Secrets. |
| Observabilidade | Logs, métricas e traces usam correlação por request_id, trace_id, session_id, tenant_id e agent_id. |
| Auditabilidade | Decisões de rota, guardrail, judge, MCP e LLM são rastreáveis. |
| Portabilidade | Execução suportada em local, Docker Compose e Kubernetes/OKE. |
| Configuração | Comportamento variável é controlado por `.env` e YAML versionado. |
## Critérios de Aceite
- [ ] Endpoint de chat completion aceita LLMRequest.
- [ ] Endpoint de embeddings aceita EmbeddingRequest.
- [ ] Profiles são resolvidos por `llm_profiles.yaml`.
- [ ] Provider/model/profile aparecem em logs e traces.
- [ ] Fallback é explícito e rastreável.
- [ ] Rate limit é aplicado por tenant/agente.
- [ ] OCI Auth funciona por ambiente.
- [ ] Modelo não autorizado é rejeitado.
- [ ] Uso de tokens é registrado.
- [ ] Erros retornam payload padronizado.
## Glossário
| Termo | Definição |
|---|---|
| Agent Platform | Plataforma composta por runtime, gateways, evaluator, templates, contratos e componentes operacionais. |
| Agent Framework | Biblioteca/core reutilizável com contratos, guardrails, judges, memória, telemetria, providers e utilitários. |
| Agent Runtime | Motor de execução de agentes baseado em LangGraph, estado, sessão, memória, checkpoints, roteamento e ciclo de vida. |
| Agent Gateway | Aplicação deployável de entrada, roteamento e orquestração entre backends/agentes. |
| Channel Gateway | Aplicação ou módulo de normalização de payloads de canais para GatewayRequest. |
| AI Gateway | Aplicação de governança, roteamento e abstração de chamadas LLM/embedding. |
| MCP Gateway | Aplicação de governança e roteamento de tools MCP. |
| Evaluator | Camada de avaliação online/offline, regressão e certificação. |
| Business Context | Conjunto de chaves canônicas de negócio: customer_key, contract_key, interaction_key, account_key, resource_key e session_key. |

View File

@@ -0,0 +1,273 @@
# SPEC-003 — Agent Gateway
## Escopo
O Agent Gateway é o ponto único de entrada da plataforma para canais e consumidores externos.
Sua responsabilidade é receber mensagens, gerenciar sessões globais, resolver o backend/agente responsável, executar roteamento, realizar handoff entre agentes/backends e encaminhar eventos SSE.
O Agent Gateway não executa inferência LLM nem embeddings. Essas capacidades pertencem ao Runtime e ao Agent Framework.
---
## Responsabilidades
### Entrada Única da Plataforma
```text
Web
WhatsApp
Voice
Teams
Slack
|
v
Agent Gateway
|
+--> Agent Backend A
|
+--> Agent Backend B
|
+--> Agent Backend C
```
### Gerenciamento de Sessões
Responsável por:
- Criação de sessões
- Recuperação de sessões
- Atualização de contexto global
- Persistência de metadados de sessão
- Correlação de requisições
Exemplo:
```json
{
"session_id": "default:telecom_contas:123",
"tenant_id": "default",
"active_backend": "telecom_contas",
"active_agent": "telecom_contas",
"turn_count": 12,
"metadata": {}
}
```
---
## Backend Routing
Resolve qual backend deve processar a mensagem.
Exemplo:
```yaml
backends:
telecom_contas:
url: http://backend-contas:8000
telecom_ofertas:
url: http://backend-ofertas:8000
```
Critérios possíveis:
- Backend padrão
- Regras YAML
- Intenção detectada
- Contexto da sessão
- Router LLM (opcional)
---
## Handoff
Permite transferência entre agentes ou backends.
Exemplo:
```text
Contas
|
+--> Ofertas
|
+--> Retenção
```
O handoff deve preservar:
- session_id
- conversation_key
- business context
- histórico da conversa
- metadados de correlação
---
## SSE Proxy
Responsável por encaminhar eventos de streaming para clientes.
### Endpoints
| Método | Endpoint |
|----------|----------|
| POST | /gateway/message |
| POST | /gateway/message/sse |
| GET | /gateway/events/{session_id} |
Eventos SSE suportados:
- connected
- workflow.started
- message.responded
- workflow.completed
- flow.end
- error
---
## Backend Discovery
Pode operar com catálogo estático ou descoberta dinâmica.
### Catálogo Estático
```yaml
backends:
telecom_contas:
url: http://contas:8000
telecom_ofertas:
url: http://ofertas:8000
```
### Descoberta Dinâmica
```yaml
service_discovery:
enabled: true
```
Capacidades:
- Registro automático
- Health check periódico
- Atualização de catálogo
- Sincronização de metadados
---
## Health e Operação
### Endpoints
| Método | Endpoint |
|----------|----------|
| GET | /health |
| GET | /ready |
| GET | /backends |
| GET | /debug/sessions |
---
## Contrato GatewayRequest
```json
{
"tenant_id": "default",
"agent_id": "telecom_contas",
"session_id": "default:telecom_contas:123",
"message": "Quero consultar minha fatura",
"business_context": {
"customer_key": "11999999999"
},
"metadata": {
"request_id": "req-001",
"trace_id": "trace-001"
}
}
```
---
## Contrato GatewayResponse
```json
{
"session_id": "default:telecom_contas:123",
"backend": "telecom_contas",
"agent": "telecom_contas",
"message": "Sua fatura está disponível.",
"metadata": {
"request_id": "req-001"
}
}
```
---
## Eventos
| Evento | Descrição |
|----------|----------|
| agent.gateway.request.received | Requisição recebida |
| agent.gateway.session.created | Sessão criada |
| agent.gateway.backend.selected | Backend selecionado |
| agent.gateway.handoff.started | Handoff iniciado |
| agent.gateway.handoff.completed | Handoff concluído |
| agent.gateway.sse.connected | Cliente SSE conectado |
| agent.gateway.request.failed | Falha de processamento |
---
## Métricas
| Métrica | Dimensões |
|----------|----------|
| gateway_requests_total | tenant, backend, agent, status |
| gateway_sessions_active | tenant |
| gateway_backend_selection_total | backend |
| gateway_handoff_total | origem, destino |
| gateway_latency_ms | backend |
| gateway_sse_connections | backend |
---
## Segurança
- Autenticação obrigatória quando configurada.
- Propagação de identidade entre gateways.
- Máscara de dados sensíveis em logs.
- Correlação por request_id, trace_id e session_id.
- Controle de acesso por tenant.
---
## Requisitos Não Funcionais
| Categoria | Requisito |
|----------|----------|
| Disponibilidade | Expor /health e /ready |
| Escalabilidade | Stateless com escala horizontal |
| Observabilidade | Logs, métricas e traces |
| Auditabilidade | Todas as decisões de roteamento rastreáveis |
| Segurança | Segredos externos e mascaramento |
| Portabilidade | Local, Docker e Kubernetes |
| Configuração | YAML e variáveis de ambiente |
---
## Critérios de Aceite
- [ ] Recebe mensagens de múltiplos canais.
- [ ] Seleciona backend corretamente.
- [ ] Mantém sessão global.
- [ ] Encaminha SSE.
- [ ] Executa handoff.
- [ ] Preserva Business Context.
- [ ] Suporta múltiplos backends.
- [ ] Permite descoberta dinâmica.
- [ ] Expõe health e readiness.
- [ ] Gera métricas e telemetria.

View File

@@ -25,8 +25,6 @@ OCI_GENAI_MODEL=openai.gpt-4.1
OCI_GENAI_API_KEY=sk-ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6
OCI_GENAI_PROJECT_OCID=
# OCI_AUTH_MODE=config_file|instance_principal|resource_principal
OCI_AUTH_MODE=config_file
# OCI SDK / signer / profiles
OCI_CONFIG_FILE=~/.oci/config
OCI_PROFILE=DEFAULT
@@ -37,9 +35,10 @@ OCI_REGION=us-chicago-1
# Persistência
###############################################################################
# Opções: memory, autonomous, mongodb
SESSION_REPOSITORY_PROVIDER=autonomous
MEMORY_REPOSITORY_PROVIDER=autonomous
CHECKPOINT_REPOSITORY_PROVIDER=autonomous
SESSION_REPOSITORY_PROVIDER=sqlite
MEMORY_REPOSITORY_PROVIDER=sqlite
CHECKPOINT_REPOSITORY_PROVIDER=sqlite
SQLITE_DB_PATH=./data/agent_framework.db
# Autonomous Database
ADB_USER=admin
@@ -60,8 +59,8 @@ ENABLE_REDIS_CACHE=false
###############################################################################
# RAG / Vector / Graph
###############################################################################
VECTOR_STORE_PROVIDER=memory
GRAPH_STORE_PROVIDER=memory
VECTOR_STORE_PROVIDER=sqlite
GRAPH_STORE_PROVIDER=sqlite
RAG_TOP_K=5
EMBEDDING_PROVIDER=mock
OCI_EMBEDDING_MODEL=cohere.embed-multilingual-v3.0
@@ -71,8 +70,9 @@ RAG_FILE_GLOBS=*.md,*.txt,*.yaml,*.yml,*.json
# Observabilidade
###############################################################################
ENABLE_LANGFUSE=true
LANGFUSE_PUBLIC_KEY=pk-lf-bd9b0c7e-2b8b-4e5b-a382-284a9b4413b3
LANGFUSE_SECRET_KEY=sk-lf-5f5cc18d-0bb5-424e-b5d0-cb3664d58c20
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=pk-lf-2f9da109-5b0f-4c78-b61d-9598ed787eba
LANGFUSE_SECRET_KEY=sk-lf-a4cb0cdd-f2ea-4468-9911-cebeb91ba944
LANGFUSE_HOST=http://localhost:3005
ENABLE_OTEL=false
OTEL_EXPORTER_OTLP_ENDPOINT=
@@ -85,7 +85,7 @@ ENABLE_LANGFUSE_OPENAI_AUTO_INSTRUMENTATION=true
# Quando true, AgentObserver publica eventos IC.*, NOC.* e GRL.* nos providers abaixo.
ENABLE_ANALYTICS=false
# Providers aceitos: oci_streaming,pubsub,noop
ANALYTICS_PROVIDERS=oci_streaming
ANALYTICS_PROVIDERS=pubsub
# Compatibilidade FIRST/TIM: pode informar AGENT_PUBSUB_TOPIC diretamente.
AGENT_PUBSUB_TOPIC=
GCP_PUBSUB_TOPIC_PATH=
@@ -150,7 +150,7 @@ MCP_TOOL_TIMEOUT_SECONDS=30
ROUTING_MODE=router
# Usage/cost accounting
USAGE_REPOSITORY_PROVIDER=autonomous
USAGE_REPOSITORY_PROVIDER=sqlite
IDENTITY_CONFIG_PATH=./config/identity.yaml
MCP_PARAMETER_MAPPING_PATH=./config/mcp_parameter_mapping.yaml

View File

@@ -74,6 +74,7 @@ logger.info("Framework channel input mode: %s", gateway.input_mode)
@app.middleware("http")
async def observability_context_middleware(request: Request, call_next):
clear_observability_context()
request_id = request.headers.get("x-request-id") or str(uuid4())
set_observability_context(
request_id=request_id,
@@ -99,6 +100,8 @@ async def observability_context_middleware(request: Request, call_next):
"duration_ms": int((time.time() - started) * 1000),
}, kind="http")
raise
finally:
clear_observability_context()
class GatewayRequest(BaseModel):
@@ -108,6 +111,31 @@ class GatewayRequest(BaseModel):
tenant_id: str | None = None
def _metadata_value(payload: dict, key: str):
metadata = payload.get("metadata")
if isinstance(metadata, dict):
return metadata.get(key)
return None
def _extract_workflow_id(payload: dict) -> str | None:
return (
payload.get("workflow_id")
or payload.get("workflowId")
or _metadata_value(payload, "workflow_id")
or _metadata_value(payload, "workflowId")
)
def _format_root_span_name(template: str | None, values: dict) -> str:
template = template or "agent.gateway_message"
try:
return template.format(**{k: v or "unknown" for k, v in values.items()})
except Exception:
logger.warning("LANGFUSE_ROOT_SPAN_NAME inválido: %s", template)
return "agent.gateway_message"
def _resolve_identity(req: GatewayRequest, msg) -> tuple[AgentIdentity, dict, BusinessContext, list[str]]:
payload = req.payload or {}
context = dict(msg.context or {})
@@ -143,9 +171,11 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
msg = await gateway.normalize(req.channel, req.payload)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
payload = req.payload or {}
identity, normalized_context, business_context, missing_identity_keys = _resolve_identity(req, msg)
agent_session_id = identity.conversation_key()
message_id = (req.payload or {}).get("message_id") or str(uuid4())
message_id = payload.get("message_id") or str(uuid4())
workflow_id = _extract_workflow_id(payload)
set_observability_context(
session_id=agent_session_id,
user_id=msg.user_id,
@@ -153,7 +183,8 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
agent_id=identity.agent_id,
channel=msg.channel,
message_id=message_id,
ura_call_id=(req.payload or {}).get("ura_call_id") or normalized_context.get("ura_call_id") or business_context.interaction_key,
workflow_id=workflow_id,
ura_call_id=payload.get("ura_call_id") or normalized_context.get("ura_call_id") or business_context.interaction_key,
)
stream = sse_hub.stream_for(agent_session_id)
@@ -209,34 +240,56 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
await sse_hub.emit(agent_session_id, "message.received", {"session_id": agent_session_id, "role": "user"}) if emit_sse else None
history = [m.model_dump(mode="json") for m in await memory.list(agent_session_id)]
trace_input = {
cms_input = {
"channel": req.channel,
"tenant_id": req.tenant_id,
"agent_id": req.agent_id,
"payload": payload,
}
trace_context = {
"text": msg.text,
"channel": msg.channel,
"channel_id": msg.channel_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"conversation_key": agent_session_id,
"workflow_id": workflow_id,
"message_id": message_id,
"business_context": business_context.model_dump(),
"identity_missing": missing_identity_keys,
}
root_span_name = _format_root_span_name(
getattr(settings, "LANGFUSE_ROOT_SPAN_NAME", "agent.gateway_message"),
{
"workflow_id": workflow_id,
"channel": msg.channel,
"agent_id": identity.agent_id,
"tenant_id": identity.tenant_id,
},
)
root_tags = ["agent-template", msg.channel, f"agent:{identity.agent_id}", f"tenant:{identity.tenant_id}"]
if workflow_id:
root_tags.append(f"workflow:{workflow_id}")
async with telemetry.span(
"agent.gateway_message",
root_span_name,
session_id=agent_session_id,
user_id=session.user_id,
channel=msg.channel,
input=trace_input,
tags=["agent-template", msg.channel, f"agent:{identity.agent_id}", f"tenant:{identity.tenant_id}"],
):
await telemetry.event("gateway.message.received", trace_input)
await sse_hub.emit(agent_session_id, "workflow.started", trace_input) if emit_sse else None
workflow_id=workflow_id,
input=cms_input,
tags=root_tags,
_root_span=True,
) as root_span:
await telemetry.event("gateway.message.received", trace_context)
await sse_hub.emit(agent_session_id, "workflow.started", trace_context) if emit_sse else None
result = await workflow.ainvoke(
{
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"session_id": agent_session_id,
"conversation_key": agent_session_id,
"workflow_id": workflow_id,
"agent_profile": normalized_context["agent_profile"],
"user_text": msg.text,
"history": history,
@@ -246,6 +299,7 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
"original_session_id": msg.session_id,
"session_id": agent_session_id,
"conversation_key": agent_session_id,
"workflow_id": workflow_id,
"user_id": session.user_id,
"channel": msg.channel,
"message_id": message_id,
@@ -256,66 +310,68 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
}
)
await checkpoints.put(agent_session_id, {"state": result, "message_id": message_id})
await sse_hub.emit(agent_session_id, "workflow.completed", {"session_id": agent_session_id, "route": result.get("route"), "intent": result.get("intent")}) if emit_sse else None
await checkpoints.put(agent_session_id, {"state": result, "message_id": message_id})
await sse_hub.emit(agent_session_id, "workflow.completed", {"session_id": agent_session_id, "route": result.get("route"), "intent": result.get("intent")}) if emit_sse else None
answer = result.get("final_answer") or result.get("answer") or ""
await memory.append(
agent_session_id,
ChatMessage(
role="assistant",
content=answer,
metadata={
answer = result.get("final_answer") or result.get("answer") or ""
await memory.append(
agent_session_id,
ChatMessage(
role="assistant",
content=answer,
metadata={
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"message_id": f"assistant-{message_id}",
"route": result.get("route"),
"intent": result.get("intent"),
"route_decision": result.get("route_decision"),
"judges": result.get("judge_results"),
},
),
)
await telemetry.event(
"gateway.message.responded",
{
"session_id": agent_session_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"message_id": f"assistant-{message_id}",
"route": result.get("route"),
"intent": result.get("intent"),
"answer_chars": len(answer),
},
)
response = ChannelResponse(
channel=msg.channel,
session_id=agent_session_id,
text=answer,
metadata={
"channel_id": msg.channel_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"original_session_id": msg.session_id,
"conversation_key": agent_session_id,
"workflow_id": workflow_id,
"message_id": message_id,
"route": result.get("route"),
"intent": result.get("intent"),
"route_decision": result.get("route_decision"),
"domain": result.get("domain"),
"mcp_tools": result.get("mcp_tools"),
"mcp_results": result.get("mcp_results"),
"business_context": business_context.model_dump(),
"identity_missing": missing_identity_keys,
"judges": result.get("judge_results"),
"guardrails": result.get("guardrail_decisions"),
},
),
)
await telemetry.event(
"gateway.message.responded",
{
"session_id": agent_session_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"route": result.get("route"),
"intent": result.get("intent"),
"answer_chars": len(answer),
},
)
response = ChannelResponse(
channel=msg.channel,
session_id=agent_session_id,
text=answer,
metadata={
"channel_id": msg.channel_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"original_session_id": msg.session_id,
"conversation_key": agent_session_id,
"message_id": message_id,
"route": result.get("route"),
"intent": result.get("intent"),
"route_decision": result.get("route_decision"),
"domain": result.get("domain"),
"mcp_tools": result.get("mcp_tools"),
"mcp_results": result.get("mcp_results"),
"business_context": business_context.model_dump(),
"identity_missing": missing_identity_keys,
"judges": result.get("judge_results"),
"guardrails": result.get("guardrail_decisions"),
},
)
rendered = await gateway.render(response)
await sse_hub.emit(agent_session_id, "message.responded", rendered) if emit_sse else None
await sse_hub.emit(agent_session_id, "flow.end", {"session_id": agent_session_id, "message_id": message_id}) if emit_sse else None
return rendered
)
rendered = await gateway.render(response)
root_span.set_output(rendered)
await sse_hub.emit(agent_session_id, "message.responded", rendered) if emit_sse else None
await sse_hub.emit(agent_session_id, "flow.end", {"session_id": agent_session_id, "message_id": message_id}) if emit_sse else None
return rendered
@app.get("/health")

View File

@@ -6,6 +6,7 @@ class AgentState(TypedDict, total=False):
agent_id: str
session_id: str
conversation_key: str
workflow_id: str
agent_profile: dict[str, Any]
user_text: str
sanitized_input: str

View File

@@ -320,41 +320,36 @@ class AgentWorkflow:
}
async def billing_agent(self, state):
async with self.langgraph_telemetry.node("billing_agent", state):
async with self.telemetry.span(
"workflow.agent.billing",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.billing.run(state)
async with self.telemetry.span(
"workflow.agent.billing",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.billing.run(state)
async def product_agent(self, state):
async with self.langgraph_telemetry.node("product_agent", state):
async with self.telemetry.span(
"workflow.agent.product",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.product.run(state)
async with self.telemetry.span(
"workflow.agent.product",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.product.run(state)
async def orders_agent(self, state):
async with self.langgraph_telemetry.node("orders_agent", state):
async with self.telemetry.span(
"workflow.agent.orders",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.orders.run(state)
async with self.telemetry.span(
"workflow.agent.orders",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.orders.run(state)
async def support_agent(self, state):
async with self.langgraph_telemetry.node("support_agent", state):
async with self.telemetry.span(
"workflow.agent.support",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.support.run(state)
async with self.telemetry.span(
"workflow.agent.support",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.support.run(state)
async def supervisor_agent(self, state):
"""Executa um ou mais agentes no modo supervisor e consolida a resposta.

View File

@@ -25,8 +25,6 @@ OCI_GENAI_MODEL=openai.gpt-4.1
OCI_GENAI_API_KEY=sk-ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6ph3FgX6
OCI_GENAI_PROJECT_OCID=
# OCI_AUTH_MODE=config_file|instance_principal|resource_principal
OCI_AUTH_MODE=config_file
# OCI SDK / signer / profiles
OCI_CONFIG_FILE=~/.oci/config
OCI_PROFILE=DEFAULT
@@ -37,9 +35,10 @@ OCI_REGION=us-chicago-1
# Persistência
###############################################################################
# Opções: memory, autonomous, mongodb
SESSION_REPOSITORY_PROVIDER=autonomous
MEMORY_REPOSITORY_PROVIDER=autonomous
CHECKPOINT_REPOSITORY_PROVIDER=autonomous
SESSION_REPOSITORY_PROVIDER=sqlite
MEMORY_REPOSITORY_PROVIDER=sqlite
CHECKPOINT_REPOSITORY_PROVIDER=sqlite
SQLITE_DB_PATH=./data/agent_framework.db
# Autonomous Database
ADB_USER=admin
@@ -60,8 +59,8 @@ ENABLE_REDIS_CACHE=false
###############################################################################
# RAG / Vector / Graph
###############################################################################
VECTOR_STORE_PROVIDER=memory
GRAPH_STORE_PROVIDER=memory
VECTOR_STORE_PROVIDER=sqlite
GRAPH_STORE_PROVIDER=sqlite
RAG_TOP_K=5
EMBEDDING_PROVIDER=mock
OCI_EMBEDDING_MODEL=cohere.embed-multilingual-v3.0
@@ -71,8 +70,9 @@ RAG_FILE_GLOBS=*.md,*.txt,*.yaml,*.yml,*.json
# Observabilidade
###############################################################################
ENABLE_LANGFUSE=true
LANGFUSE_PUBLIC_KEY=pk-lf-bd9b0c7e-2b8b-4e5b-a382-284a9b4413b3
LANGFUSE_SECRET_KEY=sk-lf-5f5cc18d-0bb5-424e-b5d0-cb3664d58c20
LANGFUSE_TRACE_MODE=compact # Opcional: verbose, compact
LANGFUSE_PUBLIC_KEY=pk-lf-2f9da109-5b0f-4c78-b61d-9598ed787eba
LANGFUSE_SECRET_KEY=sk-lf-a4cb0cdd-f2ea-4468-9911-cebeb91ba944
LANGFUSE_HOST=http://localhost:3005
ENABLE_OTEL=false
OTEL_EXPORTER_OTLP_ENDPOINT=
@@ -85,7 +85,7 @@ ENABLE_LANGFUSE_OPENAI_AUTO_INSTRUMENTATION=true
# Quando true, AgentObserver publica eventos IC.*, NOC.* e GRL.* nos providers abaixo.
ENABLE_ANALYTICS=false
# Providers aceitos: oci_streaming,pubsub,noop
ANALYTICS_PROVIDERS=oci_streaming
ANALYTICS_PROVIDERS=pubsub
# Compatibilidade FIRST/TIM: pode informar AGENT_PUBSUB_TOPIC diretamente.
AGENT_PUBSUB_TOPIC=
GCP_PUBSUB_TOPIC_PATH=
@@ -122,9 +122,6 @@ PROMPT_POLICY_PATH=./config/prompt_policy.yaml
# Gateway de canais
###############################################################################
DEFAULT_CHANNEL=web
# embedded = backend may parse simple/native channel payloads.
# external = backend only accepts GatewayRequest normalized by an external Channel Gateway.
FRAMEWORK_CHANNEL_INPUT_MODE=embedded
ENABLE_VOICE_ADAPTER=true
ENABLE_WHATSAPP_ADAPTER=true
ENABLE_TEXT_ADAPTER=true
@@ -150,7 +147,7 @@ MCP_TOOL_TIMEOUT_SECONDS=30
ROUTING_MODE=router
# Usage/cost accounting
USAGE_REPOSITORY_PROVIDER=autonomous
USAGE_REPOSITORY_PROVIDER=sqlite
IDENTITY_CONFIG_PATH=./config/identity.yaml
MCP_PARAMETER_MAPPING_PATH=./config/mcp_parameter_mapping.yaml

View File

@@ -4,7 +4,7 @@ import logging
from uuid import uuid4
import time
from fastapi import FastAPI, Request
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
@@ -52,7 +52,7 @@ summary_memory = create_conversation_summary_memory(settings, message_history=me
sessions = create_session_repository(settings)
checkpoints = create_checkpoint_repository(settings)
cache = create_cache(settings, telemetry=telemetry)
gateway = ChannelGateway()
gateway = ChannelGateway(input_mode=settings.FRAMEWORK_CHANNEL_INPUT_MODE)
analytics = create_analytics_publisher(settings)
observer = TelemetryBackedAgentObserver(telemetry=telemetry)
configure_global_observer({
@@ -70,9 +70,11 @@ logger.info("LLM provider carregado: %s", llm.__class__.__name__)
logger.info("Langfuse habilitado: %s host=%s", telemetry.is_enabled(), settings.LANGFUSE_HOST)
logger.info("Analytics habilitado: %s providers=%s", getattr(settings, "ENABLE_ANALYTICS", False), getattr(settings, "ANALYTICS_PROVIDERS", ""))
logger.info("Agentes disponíveis: %s", [p.agent_id for p in agent_profiles.list_profiles()])
logger.info("Framework channel input mode: %s", gateway.input_mode)
@app.middleware("http")
async def observability_context_middleware(request: Request, call_next):
clear_observability_context()
request_id = request.headers.get("x-request-id") or str(uuid4())
set_observability_context(
request_id=request_id,
@@ -98,6 +100,8 @@ async def observability_context_middleware(request: Request, call_next):
"duration_ms": int((time.time() - started) * 1000),
}, kind="http")
raise
finally:
clear_observability_context()
class GatewayRequest(BaseModel):
@@ -107,6 +111,31 @@ class GatewayRequest(BaseModel):
tenant_id: str | None = None
def _metadata_value(payload: dict, key: str):
metadata = payload.get("metadata")
if isinstance(metadata, dict):
return metadata.get(key)
return None
def _extract_workflow_id(payload: dict) -> str | None:
return (
payload.get("workflow_id")
or payload.get("workflowId")
or _metadata_value(payload, "workflow_id")
or _metadata_value(payload, "workflowId")
)
def _format_root_span_name(template: str | None, values: dict) -> str:
template = template or "agent.gateway_message"
try:
return template.format(**{k: v or "unknown" for k, v in values.items()})
except Exception:
logger.warning("LANGFUSE_ROOT_SPAN_NAME inválido: %s", template)
return "agent.gateway_message"
def _resolve_identity(req: GatewayRequest, msg) -> tuple[AgentIdentity, dict, BusinessContext, list[str]]:
payload = req.payload or {}
context = dict(msg.context or {})
@@ -138,10 +167,15 @@ def _resolve_identity(req: GatewayRequest, msg) -> tuple[AgentIdentity, dict, Bu
async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False) -> dict:
msg = await gateway.normalize(req.channel, req.payload)
try:
msg = await gateway.normalize(req.channel, req.payload)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
payload = req.payload or {}
identity, normalized_context, business_context, missing_identity_keys = _resolve_identity(req, msg)
agent_session_id = identity.conversation_key()
message_id = (req.payload or {}).get("message_id") or str(uuid4())
message_id = payload.get("message_id") or str(uuid4())
workflow_id = _extract_workflow_id(payload)
set_observability_context(
session_id=agent_session_id,
user_id=msg.user_id,
@@ -149,7 +183,8 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
agent_id=identity.agent_id,
channel=msg.channel,
message_id=message_id,
ura_call_id=(req.payload or {}).get("ura_call_id") or normalized_context.get("ura_call_id") or business_context.interaction_key,
workflow_id=workflow_id,
ura_call_id=payload.get("ura_call_id") or normalized_context.get("ura_call_id") or business_context.interaction_key,
)
stream = sse_hub.stream_for(agent_session_id)
@@ -205,34 +240,56 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
await sse_hub.emit(agent_session_id, "message.received", {"session_id": agent_session_id, "role": "user"}) if emit_sse else None
history = [m.model_dump(mode="json") for m in await memory.list(agent_session_id)]
trace_input = {
cms_input = {
"channel": req.channel,
"tenant_id": req.tenant_id,
"agent_id": req.agent_id,
"payload": payload,
}
trace_context = {
"text": msg.text,
"channel": msg.channel,
"channel_id": msg.channel_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"conversation_key": agent_session_id,
"workflow_id": workflow_id,
"message_id": message_id,
"business_context": business_context.model_dump(),
"identity_missing": missing_identity_keys,
}
root_span_name = _format_root_span_name(
getattr(settings, "LANGFUSE_ROOT_SPAN_NAME", "agent.gateway_message"),
{
"workflow_id": workflow_id,
"channel": msg.channel,
"agent_id": identity.agent_id,
"tenant_id": identity.tenant_id,
},
)
root_tags = ["agent-template", msg.channel, f"agent:{identity.agent_id}", f"tenant:{identity.tenant_id}"]
if workflow_id:
root_tags.append(f"workflow:{workflow_id}")
async with telemetry.span(
"agent.gateway_message",
root_span_name,
session_id=agent_session_id,
user_id=session.user_id,
channel=msg.channel,
input=trace_input,
tags=["agent-template", msg.channel, f"agent:{identity.agent_id}", f"tenant:{identity.tenant_id}"],
):
await telemetry.event("gateway.message.received", trace_input)
await sse_hub.emit(agent_session_id, "workflow.started", trace_input) if emit_sse else None
workflow_id=workflow_id,
input=cms_input,
tags=root_tags,
_root_span=True,
) as root_span:
await telemetry.event("gateway.message.received", trace_context)
await sse_hub.emit(agent_session_id, "workflow.started", trace_context) if emit_sse else None
result = await workflow.ainvoke(
{
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"session_id": agent_session_id,
"conversation_key": agent_session_id,
"workflow_id": workflow_id,
"agent_profile": normalized_context["agent_profile"],
"user_text": msg.text,
"history": history,
@@ -242,6 +299,7 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
"original_session_id": msg.session_id,
"session_id": agent_session_id,
"conversation_key": agent_session_id,
"workflow_id": workflow_id,
"user_id": session.user_id,
"channel": msg.channel,
"message_id": message_id,
@@ -252,66 +310,68 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False)
}
)
await checkpoints.put(agent_session_id, {"state": result, "message_id": message_id})
await sse_hub.emit(agent_session_id, "workflow.completed", {"session_id": agent_session_id, "route": result.get("route"), "intent": result.get("intent")}) if emit_sse else None
await checkpoints.put(agent_session_id, {"state": result, "message_id": message_id})
await sse_hub.emit(agent_session_id, "workflow.completed", {"session_id": agent_session_id, "route": result.get("route"), "intent": result.get("intent")}) if emit_sse else None
answer = result.get("final_answer") or result.get("answer") or ""
await memory.append(
agent_session_id,
ChatMessage(
role="assistant",
content=answer,
metadata={
answer = result.get("final_answer") or result.get("answer") or ""
await memory.append(
agent_session_id,
ChatMessage(
role="assistant",
content=answer,
metadata={
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"message_id": f"assistant-{message_id}",
"route": result.get("route"),
"intent": result.get("intent"),
"route_decision": result.get("route_decision"),
"judges": result.get("judge_results"),
},
),
)
await telemetry.event(
"gateway.message.responded",
{
"session_id": agent_session_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"message_id": f"assistant-{message_id}",
"route": result.get("route"),
"intent": result.get("intent"),
"answer_chars": len(answer),
},
)
response = ChannelResponse(
channel=msg.channel,
session_id=agent_session_id,
text=answer,
metadata={
"channel_id": msg.channel_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"original_session_id": msg.session_id,
"conversation_key": agent_session_id,
"workflow_id": workflow_id,
"message_id": message_id,
"route": result.get("route"),
"intent": result.get("intent"),
"route_decision": result.get("route_decision"),
"domain": result.get("domain"),
"mcp_tools": result.get("mcp_tools"),
"mcp_results": result.get("mcp_results"),
"business_context": business_context.model_dump(),
"identity_missing": missing_identity_keys,
"judges": result.get("judge_results"),
"guardrails": result.get("guardrail_decisions"),
},
),
)
await telemetry.event(
"gateway.message.responded",
{
"session_id": agent_session_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"route": result.get("route"),
"intent": result.get("intent"),
"answer_chars": len(answer),
},
)
response = ChannelResponse(
channel=msg.channel,
session_id=agent_session_id,
text=answer,
metadata={
"channel_id": msg.channel_id,
"tenant_id": identity.tenant_id,
"agent_id": identity.agent_id,
"original_session_id": msg.session_id,
"conversation_key": agent_session_id,
"message_id": message_id,
"route": result.get("route"),
"intent": result.get("intent"),
"route_decision": result.get("route_decision"),
"domain": result.get("domain"),
"mcp_tools": result.get("mcp_tools"),
"mcp_results": result.get("mcp_results"),
"business_context": business_context.model_dump(),
"identity_missing": missing_identity_keys,
"judges": result.get("judge_results"),
"guardrails": result.get("guardrail_decisions"),
},
)
rendered = await gateway.render(response)
await sse_hub.emit(agent_session_id, "message.responded", rendered) if emit_sse else None
await sse_hub.emit(agent_session_id, "flow.end", {"session_id": agent_session_id, "message_id": message_id}) if emit_sse else None
return rendered
)
rendered = await gateway.render(response)
root_span.set_output(rendered)
await sse_hub.emit(agent_session_id, "message.responded", rendered) if emit_sse else None
await sse_hub.emit(agent_session_id, "flow.end", {"session_id": agent_session_id, "message_id": message_id}) if emit_sse else None
return rendered
@app.get("/health")
@@ -331,6 +391,8 @@ async def health():
"usage_repository": settings.USAGE_REPOSITORY_PROVIDER,
"identity_config_path": settings.IDENTITY_CONFIG_PATH,
"mcp_parameter_mapping_path": settings.MCP_PARAMETER_MAPPING_PATH,
"framework_channel_input_mode": settings.FRAMEWORK_CHANNEL_INPUT_MODE,
"legacy_channel_gateway_mode": settings.CHANNEL_GATEWAY_MODE,
}
@@ -354,6 +416,8 @@ async def debug_env():
"AGENTS_CONFIG_PATH": settings.AGENTS_CONFIG_PATH,
"ROUTING_CONFIG_PATH": settings.ROUTING_CONFIG_PATH,
"ROUTING_MODE": settings.ROUTING_MODE,
"FRAMEWORK_CHANNEL_INPUT_MODE": settings.FRAMEWORK_CHANNEL_INPUT_MODE,
"CHANNEL_GATEWAY_MODE": settings.CHANNEL_GATEWAY_MODE,
}

View File

@@ -6,6 +6,7 @@ class AgentState(TypedDict, total=False):
agent_id: str
session_id: str
conversation_key: str
workflow_id: str
agent_profile: dict[str, Any]
user_text: str
sanitized_input: str

View File

@@ -187,6 +187,16 @@ class AgentWorkflow:
input=state.get("user_text"),
):
history_texts = [m.get("content", "") for m in state.get("history", [])]
await self.observer.emit_grl(
"001",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "input",
},
component="workflow.input_guardrails.start",
)
sanitized, decisions = await self.guardrails.run_input(
state["user_text"],
{
@@ -199,6 +209,19 @@ class AgentWorkflow:
)
for _decision in decisions:
await self.guardrail_telemetry.evaluated("input", _decision)
await self.observer.emit_grl(
"002" if _decision.allowed else "004",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "input",
"rail_code": getattr(_decision, "code", None),
"allowed": bool(_decision.allowed),
"reason": getattr(_decision, "reason", None),
},
component="workflow.input_guardrails.decision",
)
if not _decision.allowed:
await self.guardrail_telemetry.blocked("input", _decision)
await self.telemetry.event(
@@ -210,6 +233,18 @@ class AgentWorkflow:
"decisions": [d.model_dump() for d in decisions],
},
)
await self.observer.emit_grl(
"009",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "input",
"blocked": any(not d.allowed for d in decisions),
"decision_count": len(decisions),
},
component="workflow.input_guardrails.final",
)
if any(not d.allowed for d in decisions):
return {
"sanitized_input": sanitized,
@@ -285,41 +320,36 @@ class AgentWorkflow:
}
async def billing_agent(self, state):
async with self.langgraph_telemetry.node("billing_agent", state):
async with self.telemetry.span(
"workflow.agent.billing",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.billing.run(state)
async with self.telemetry.span(
"workflow.agent.billing",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.billing.run(state)
async def product_agent(self, state):
async with self.langgraph_telemetry.node("product_agent", state):
async with self.telemetry.span(
"workflow.agent.product",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.product.run(state)
async with self.telemetry.span(
"workflow.agent.product",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.product.run(state)
async def orders_agent(self, state):
async with self.langgraph_telemetry.node("orders_agent", state):
async with self.telemetry.span(
"workflow.agent.orders",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.orders.run(state)
async with self.telemetry.span(
"workflow.agent.orders",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.orders.run(state)
async def support_agent(self, state):
async with self.langgraph_telemetry.node("support_agent", state):
async with self.telemetry.span(
"workflow.agent.support",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.support.run(state)
async with self.telemetry.span(
"workflow.agent.support",
session_id=state.get("conversation_key") or state.get("session_id"),
input={"intent": state.get("intent")},
):
return await self.support.run(state)
async def supervisor_agent(self, state):
"""Executa um ou mais agentes no modo supervisor e consolida a resposta.
@@ -419,6 +449,21 @@ class AgentWorkflow:
},
)
await self.observer.emit_ic(
"IC.OUTPUT_SUPERVISOR_COMPLETED",
{
"session_id": context["session_id"],
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"route": state.get("route"),
"intent": state.get("intent"),
"action": action,
"approved": decision.approved,
"result_count": len(decision.results),
},
component="workflow.output_supervisor",
)
if decision.action in {RailAction.ALLOW, RailAction.SANITIZE, RailAction.OBSERVE}:
final_answer = decision.candidate
elif decision.action == RailAction.HANDOVER:
@@ -457,11 +502,36 @@ class AgentWorkflow:
session_id=state.get("conversation_key") or state.get("session_id"),
input=state.get("answer"),
):
await self.observer.emit_grl(
"001",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "output",
"route": state.get("route"),
"intent": state.get("intent"),
},
component="workflow.output_guardrails.start",
)
final, decisions = await self.guardrails.run_output(
state["answer"], state.get("context", {})
)
for _decision in decisions:
await self.guardrail_telemetry.evaluated("output", _decision)
await self.observer.emit_grl(
"002" if _decision.allowed else "004",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "output",
"rail_code": getattr(_decision, "code", None),
"allowed": bool(_decision.allowed),
"reason": getattr(_decision, "reason", None),
},
component="workflow.output_guardrails.decision",
)
if not _decision.allowed:
await self.guardrail_telemetry.blocked("output", _decision)
await self.telemetry.event(
@@ -473,6 +543,18 @@ class AgentWorkflow:
"decisions": [d.model_dump() for d in decisions],
},
)
await self.observer.emit_grl(
"009",
{
"session_id": state.get("conversation_key") or state.get("session_id"),
"tenant_id": state.get("tenant_id"),
"agent_id": state.get("agent_id"),
"phase": "output",
"blocked": any(not d.allowed for d in decisions),
"decision_count": len(decisions),
},
component="workflow.output_guardrails.final",
)
return {
"final_answer": final,
"guardrail_decisions": state.get("guardrail_decisions", [])

View File

@@ -0,0 +1,194 @@
from __future__ import annotations
import pytest
from agent_framework.observability.context import clear_observability_context
from agent_framework.observability.telemetry import Telemetry
class Settings:
ENABLE_LANGFUSE = False
ENABLE_OTEL = False
LANGFUSE_TRACE_MODE = "compact"
class FakeObservation:
_next_id = 1
def __init__(self, kwargs):
self.kwargs = kwargs
self.id = f"obs-{FakeObservation._next_id}"
self.trace_id = "trace-123"
FakeObservation._next_id += 1
self.updates = []
self.trace_updates = []
self.trace_io_updates = []
def update(self, **kwargs):
self.updates.append(kwargs)
def update_trace(self, **kwargs):
self.trace_updates.append(kwargs)
def set_trace_io(self, **kwargs):
self.trace_io_updates.append(kwargs)
class FakeContextManager:
def __init__(self, observation):
self.observation = observation
def __enter__(self):
return self.observation
def __exit__(self, exc_type, exc, tb):
return False
class FakePropagationContext:
def __init__(self, owner, kwargs):
self.owner = owner
self.kwargs = kwargs
def __enter__(self):
self.owner.propagations.append(self.kwargs)
def __exit__(self, exc_type, exc, tb):
return False
class FakeLangfuse:
def __init__(self, *, legacy_api: bool = False):
self.observations = []
self.propagations = []
self.flush_count = 0
self.api = FakeApi() if legacy_api else None
def start_as_current_observation(self, **kwargs):
observation = FakeObservation(kwargs)
self.observations.append(observation)
return FakeContextManager(observation)
def propagate_attributes(self, **kwargs):
return FakePropagationContext(self, kwargs)
def flush(self):
self.flush_count += 1
class FakeIngestionResponse:
errors = []
successes = []
class FakeIngestion:
def __init__(self):
self.batches = []
def batch(self, *, batch, metadata=None):
self.batches.append({"batch": batch, "metadata": metadata})
return FakeIngestionResponse()
class FakeApi:
def __init__(self):
self.ingestion = FakeIngestion()
def telemetry_with_fake_langfuse(*, legacy_api: bool = False):
FakeObservation._next_id = 1
telemetry = Telemetry(Settings())
telemetry.enabled = True
telemetry.langfuse = FakeLangfuse(legacy_api=legacy_api)
return telemetry
@pytest.mark.asyncio
async def test_compact_keeps_root_output_and_shows_only_aga_noc_events():
clear_observability_context()
telemetry = telemetry_with_fake_langfuse()
async with telemetry.span("agent.gateway_message", session_id="s1", input={"request": "cms"}, _root_span=True) as span:
await telemetry.event("IC.INTERNAL", {"step": "hidden"}, kind="ic")
await telemetry.event("NOC.001", {"step": "visible"}, kind="noc")
await telemetry.event("AGA.010", {"step": "visible"}, kind="ic")
span.set_output({"answer": "ok"})
names = [obs.kwargs["name"] for obs in telemetry.langfuse.observations]
assert names == ["agent.gateway_message", "NOC.001", "AGA.010"]
root = telemetry.langfuse.observations[0]
assert root.updates[-1]["input"] == {"request": "cms"}
assert root.updates[-1]["output"] == {"answer": "ok"}
assert root.trace_io_updates[-1] == {"input": {"request": "cms"}, "output": {"answer": "ok"}}
assert telemetry.langfuse.observations[1].kwargs.get("trace_context") is None
assert telemetry.langfuse.observations[2].kwargs.get("trace_context") is None
assert telemetry.langfuse.propagations[-1]["trace_name"] == "agent.gateway_message"
aggregated = root.updates[-1]["metadata"]["aggregated_events"]
assert [event["name"] for event in aggregated] == ["IC.INTERNAL", "NOC.001", "AGA.010"]
@pytest.mark.asyncio
async def test_compact_generation_records_io_model_parameters_and_usage_details():
clear_observability_context()
telemetry = telemetry_with_fake_langfuse()
async with telemetry.span("agent.gateway_message", session_id="s1", input={"request": "cms"}, _root_span=True):
async with telemetry.generation_span(
name="llm.test",
model="test-model",
input=[{"role": "user", "content": "ping"}],
metadata={"profile_name": "test"},
model_parameters={"temperature": 0.2, "max_tokens": 100},
) as generation:
generation.set_output("pong")
generation.set_usage({"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2, "cost_usd": 0.01})
generation = telemetry.langfuse.observations[1]
assert generation.kwargs["name"] == "llm.test"
assert generation.kwargs["as_type"] == "generation"
assert generation.kwargs["input"] == [{"role": "user", "content": "ping"}]
assert generation.kwargs["model"] == "test-model"
assert generation.kwargs["model_parameters"] == {"temperature": 0.2, "max_tokens": 100}
assert "usage" not in generation.kwargs
assert "usage_details" not in generation.kwargs
assert generation.updates[-1]["input"] == [{"role": "user", "content": "ping"}]
assert generation.updates[-1]["output"] == "pong"
assert generation.updates[-1]["usage_details"] == {"input": 1, "output": 1}
assert generation.updates[-1]["cost_details"] == {"total": 0.01}
assert generation.kwargs.get("trace_context") is None
@pytest.mark.asyncio
async def test_legacy_io_fallback_updates_same_root_and_generation_observations():
clear_observability_context()
telemetry = telemetry_with_fake_langfuse(legacy_api=True)
async with telemetry.span("agent.gateway_message", session_id="s1", input={"request": "cms"}, _root_span=True) as root:
async with telemetry.generation_span(
name="llm.test",
model="test-model",
input=[{"role": "user", "content": "ping"}],
model_parameters={"temperature": 0.2},
) as generation:
generation.set_output("pong")
generation.set_usage({"prompt_tokens": 2, "completion_tokens": 3, "total_tokens": 5})
root.set_output({"answer": "ok"})
batches = telemetry.langfuse.api.ingestion.batches
assert [event.type for item in batches for event in item["batch"]] == ["generation-update", "span-update"]
generation_event = batches[0]["batch"][0]
assert generation_event.body.id == "obs-2"
assert generation_event.body.trace_id == "trace-123"
assert generation_event.body.input == [{"role": "user", "content": "ping"}]
assert generation_event.body.output == "pong"
assert generation_event.body.usage_details == {"input": 2, "output": 3}
root_event = batches[1]["batch"][0]
assert root_event.body.id == "obs-1"
assert root_event.body.trace_id == "trace-123"
assert root_event.body.input == {"request": "cms"}
assert root_event.body.output == {"answer": "ok"}
assert len(telemetry.langfuse.observations) == 2