biatoolkit 1.3.4

Biblioteca para desenvolvedores que utilizam o BiaAgentBuilder

Bia Toolkit (biatoolkit)

Toolkit Python para facilitar o desenvolvimento e teste de MCP Servers (Model Context Protocol) integrados ao Bia Agent Builder (AWS Bedrock AgentCore).

Este repositório entrega dois grandes blocos:

1. SDK para MCP

• BiaClient: cliente para chamar um MCP Server (ListTools / CallTool)
• BiaUtil: utilitário para MCP Server ler headers do runtime e parâmetros/segredos

2. DevTools (biatoolkit.devtools)

• cli_validate: validação estática do bundle (estrutura, Dockerfile, entrypoint etc.)
• cli_smoke: validação runtime (docker build/run + ListTools MCP)

---

Sumário

• Instalação
• Conceitos rápidos
• BiaClient (consumir um MCP Server)
• BiaUtil (usar dentro do MCP Server)
• Headers do Runtime (AgentCore)
• Parâmetros e Segredos (env e SSM)
• Integração Sankhya
  • load_view (recomendado)
  • REST v1 por path (recomendado)
  • Sankhya.Call (compatibilidade / uso genérico)
  • Estratégias de sessão (debug)
  • Configurações via env
• Configurações do Toolkit (env)
• DevTools CLI
  • Validação estática: cli_validate
  • Runtime smoke: cli_smoke
• Troubleshooting
• Licença

---

Instalação

pip install mcp biatoolkit

---

Conceitos rápidos

O que é um MCP Server?

Um servidor MCP expõe tools (funções) que podem ser listadas e executadas via protocolo MCP.

Onde roda?

• Local: FastMCP + transport="streamable-http"
• Produção: Bia Agent Builder (AWS Bedrock AgentCore)

---

BiaClient (consumir um MCP Server)

O BiaClient é um cliente assíncrono que abstrai:

• conexão streamable-http
• criação/initialize de sessão MCP
• chamadas list_tools() e call_tool()

Criando um cliente e listando tools

import asyncio
from biatoolkit.basic_client import BiaClient

async def main():
    client = BiaClient("http://127.0.0.1:8000/mcp")
    tools = await client.list_tools()

    # o retorno é o objeto retornado pelo mcp.ClientSession (list_tools)
    # normalmente contém tools com name/description/schema dependendo do server
    for t in tools.tools:
        print(t.name, "-", t.description)

asyncio.run(main())

Executando uma tool (CallTool)

import asyncio
from biatoolkit.basic_client import BiaClient

async def main():
    client = BiaClient("http://127.0.0.1:8000/mcp")
    result = await client.call_tool("minha_tool", {"x": 1})

    # o shape do result depende do MCP server
    print(result)

asyncio.run(main())

Passando headers (simular runtime do AgentCore)

from biatoolkit.basic_client import BiaClient

headers = {
  "X-Amzn-Bedrock-AgentCore-Runtime-Custom-current-host": "https://meu.erp.sankhya.com.br",
  "X-Amzn-Bedrock-AgentCore-Runtime-Custom-user-email": "[e-mail]",
  "X-Amzn-Bedrock-AgentCore-Runtime-Custom-jsessionid": "JSESSIONID-ABC",
  "X-Amzn-Bedrock-AgentCore-Runtime-Custom-organization-id": "123",
  "Content-Type": "application/json",
}

client = BiaClient("http://127.0.0.1:8000/mcp", headers=headers)

Em produção, o runtime AgentCore controla quais headers são repassados.

---

BiaUtil (usar dentro do MCP Server)

O BiaUtil é usado dentro do MCP Server para:

• ler headers do runtime do AgentCore
• recuperar parâmetros/segredos de forma segura (env -> SSM fallback)

Exemplo (ler header)

from mcp.server.fastmcp import FastMCP
from biatoolkit.util import BiaUtil

mcp = FastMCP(host="0.0.0.0", stateless_http=True)

@mcp.tool()
def whoami() -> str:
    util = BiaUtil(mcp)
    h = util.get_header()
    return f"user_email={h.user_email} org={h.organization_id} host={h.current_host}"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

---

Headers do Runtime (AgentCore)

O BiaUtil.get_header() retorna um objeto Header com os campos:

Campo	Tipo	Origem
current_host	str|None	header ...-current-host
user_email	str|None	header ...-user-email
jwt_token	str|None	header ...-jwt-token
jsessionid	str|None	header ...-jsessionid
organization_id	int	header ...-organization-id (fallback 0)
codparc	int	header ...-codparc (fallback 0)
iam_user_id	int	header ...-iam-user-id (fallback 0)
gateway_token	str|None	header ...-gateway-token

Header prefix

O toolkit trabalha com um prefixo base (default):

• x-amzn-bedrock-agentcore-runtime-custom

Exemplo de header efetivo (case-insensitive em HTTP):

• X-Amzn-Bedrock-AgentCore-Runtime-Custom-user-email

---

Parâmetros e Segredos (env e SSM)

O BiaUtil.get_parameter("NOME") resolve na ordem:

1. Variável de ambiente
2. AWS SSM Parameter Store (fallback)

Como o SSM é resolvido

Para buscar no SSM, o toolkit usa um prefixo vindo do header:

• {HEADER_PREFIX}-prefix

Exemplo (nome efetivo do header):

• X-Amzn-Bedrock-AgentCore-Runtime-Custom-prefix: /bia/agentbuilder/segredos

Então o toolkit busca no SSM:

• Name = "{prefix}/{parameter_name}" com WithDecryption=True

Se o header ...-prefix não existir, o toolkit não consulta SSM e retorna None.

---

Integração Sankhya

A integração Sankhya vive em biatoolkit.sankhya_call e tem como foco:

• permitir chamadas legadas (/mge/service.sbr) e REST (/api/v1/..., /v1/...)
• usar JSESSIONID do header do runtime (quando disponível) ou valor explícito
• suportar retries/timeouts via variáveis de ambiente

Origem da URL final:

• base_url: vem de base_url explícito ou de current_host no header do runtime.
• endpoint/path: vem de quem chama (service_path ou url).
• o toolkit monta a URL final, querystring e autenticação.

load_view (recomendado)

load_view é um helper para:

• CRUDServiceProvider.loadView

Ele monta o payload e querystring automaticamente.

from mcp.server.fastmcp import FastMCP
from biatoolkit.sankhya_call import Sankhya

mcp = FastMCP(host="0.0.0.0", stateless_http=True)

@mcp.tool()
def recomendacoes() -> dict:
    sk = Sankhya(mcp=mcp)

    # base_url pode ser omitido se o header current_host existir no runtime
    return sk.load_view(
        view_name="BIA_VW_MB_RULES",
        where_sql="CODPROD_A = 123",
        fields="*",
    )

Parâmetros principais:

• view_name (obrigatório)
• where_sql (obrigatório)
• fields (opcional, default "*")
• jsessionid (opcional; se None tenta extrair do header do runtime)
• base_url (opcional; se None tenta usar current_host do header)
• url (opcional override total)
• extra_headers (opcional)

REST v1 por path (recomendado)

Para REST v1, o consumidor deve informar apenas o path do serviço.

from biatoolkit.sankhya_call import Sankhya

sk = Sankhya(mcp=mcp)

out = sk.call_json(
  method="GET",
  service_path="/api/v1/financeiros/receitas",
  query="pagina=1&tamanho=20",
  # base_url pode ser omitido: toolkit resolve via header.current_host
)

Comportamento padrão para REST v1:

• se outputType não estiver na query, o toolkit inclui outputType=json.
• sessão padrão: mgeSession truncado na query + Cookie: JSESSIONID=<completo>.
• se base_url não for passado, tenta resolver via current_host do runtime.

Sankhya.Call (compatibilidade / uso genérico)

Existe um método estático Sankhya.Call(...) para compatibilidade com scaffolds legados e uso genérico.

from biatoolkit.sankhya_call import Sankhya

out = Sankhya.Call(
  method="GET",
  service_path="/api/v1/financeiros/receitas",
  query="pagina=1&tamanho=20",
)

Parâmetros principais:

• jsessionID (opcional; se None tenta extrair do header do runtime quando mcp for informado)
• payload (opcional; body da requisição em dict)
• mcp (opcional; usado para resolver dados do runtime, como jsessionID)
• url (opcional; URL completa ou relativa)
• service_path (opcional; path do serviço, ex: /api/v1/financeiros/receitas)
• base_url (opcional; base da URL quando url não for informado)
• query (opcional; querystring, ex: serviceName=...&outputType=json)
• method (opcional, default "POST"; aceita "POST" ou "GET")
• extra_headers (opcional; headers adicionais)
• session_mode (opcional; estratégia de mgeSession/Cookie)
• include_output_type_json (opcional; força inclusão de outputType=json)

Se estiver rodando dentro de MCP Server, você pode omitir jsessionID e passar mcp=... para extrair do header.

Estratégias de sessão (debug)

Valores aceitos para session_mode:

Valor	mgeSession na query	Cookie JSESSIONID
truncated_query_and_cookie (default)	truncado (antes do .)	sim
full_query_and_cookie	completo	sim
truncated_query_only	truncado (antes do .)	não
full_query_only	completo	não
cookie_only	não envia	sim

Exemplo para depuração:

out = Sankhya.Call(
  method="GET",
  service_path="/api/v1/financeiros/receitas",
  query="pagina=1",
  session_mode="full_query_only",
)

---

Configurações via env (Sankhya)

O toolkit lê as seguintes variáveis (com defaults no código):

Variável	Default	Descrição
SANKHYA_TIMEOUT_CONNECT	3.05	timeout de conexão
SANKHYA_TIMEOUT_READ	12.0	timeout de leitura
SANKHYA_RETRIES_TOTAL	3	tentativas em falha
SANKHYA_RETRY_BACKOFF	0.5	backoff entre tentativas
SANKHYA_VERIFY_SSL	1	valida SSL (1/true/yes/on)

---

Configurações do Toolkit (env)

O toolkit expõe BiaToolkitSettings.from_env() com:

Variável	Default	Descrição
BIATOOLKIT_HEADER_PREFIX	x-amzn-bedrock-agentcore-runtime-custom	prefixo base de headers
BIATOOLKIT_AWS_REGION	sa-east-1	região para AWS SSM
BIATOOLKIT_CLIENT_TIMEOUT_SECONDS	120	timeout do cliente MCP

---

DevTools CLI

Os DevTools validam bundles de MCP Server antes do deploy no AgentCore.

Eles ficam em:

• biatoolkit.devtools

Validação estática: cli_validate

Valida estrutura e consistência do bundle (fase 1).

python -m biatoolkit.devtools.cli_validate --path ./meu_bundle

Parâmetros:

Flag	Obrigatório	Descrição
--path	✅	pasta do bundle
--entry-file	❌	arquivo Python que inicia o server (ex: app.py)
--entry-module	❌	módulo para python -m (ex: sales_agent_murilo)
--require-entry	❌	exige informar --entry-file ou --entry-module
--allow-hyphen-module	❌	transforma nome inválido (com hífen) de ERROR para WARN
--verbose	❌	logs detalhados

Runtime smoke: cli_smoke

Executa validação runtime (fase 2):

• docker build
• docker run detached
• valida container vivo por X segundos
• healthcheck HTTP (opcional)
• MCP ListTools (opcional, mas recomendado)

python -m biatoolkit.devtools.cli_smoke \
  --path ./meu_bundle \
  --port 8000 \
  --mcp-path /mcp

Parâmetros:

Flag	Obrigatório	Descrição
--path	✅	pasta do bundle
--port	❌	porta mapeada host:container (default 8000)
--mcp-path	❌	path MCP (default /mcp)
--mcp-url	❌	URL completa do MCP (override; ignora --port/--mcp-path)
--skip-mcp-list-tools	❌	pula o ListTools (debug)
--health-path	❌	path HTTP para healthcheck (ex: /health)
--run-seconds	❌	tempo que o container deve ficar vivo (default 8)
--build-timeout-sec	❌	timeout do build (default 600)
--start-timeout-sec	❌	timeout do start (default 60)
--env KEY=VALUE	❌	injeta env no container (pode repetir)
--tag	❌	tag da imagem docker
--skip-static	❌	pula validação estática
--keep-image	❌	não remove imagem
--keep-container	❌	não remove container
--verbose	❌	logs de progresso
--show-logs	❌	tail de logs no sucesso

Próxima evolução planejada do cli_smoke: CallTool automático (Item 4).

---

Troubleshooting

1) Docker daemon indisponível

• Garanta que Docker está instalado e rodando.
• O smoke test detecta e reporta esse cenário.

2) Container sobe e morre rápido

• Verifique entrypoint no Dockerfile.
• Rode com --keep-container e --show-logs para diagnóstico.

3) MCP ListTools falha (406 / handshake)

• Use o BiaClient (já é o padrão do smoke).
• Confirme o endpoint e o path (--mcp-url / --mcp-path).

4) SSM não retorna segredo

• Confirme se o header ...-prefix existe no runtime.
• Confirme a região AWS (BIATOOLKIT_AWS_REGION).
• Confirme permissão IAM para ssm:GetParameter.

---

Licença

Defina aqui o modelo de licença (ex: MIT / Apache-2.0 / Proprietária).