biatoolkit 1.0.4

Biblioteca para desenvolvedores que utilizam o BiaAgentBuilder

Bia Toolkit

Biblioteca Python para facilitar o desenvolvimento de servidores MCP integrados ao Bia Agent Builder.

Principais classes da biblioteca

Nesta seção você encontrará uma breve descrição de cada classe da biblioteca Bia Toolkit.

BiaClient

A classe BiaClient tem o objetivo de criar um cliente HTTP para comunicação com servidores MCP (Model Context Protocol).

Principais métodos:

• list_tools: Lista todas as ferramentas disponíveis no servidor MCP.
• call_tool: Executa uma ferramenta específica no servidor MCP.

BiaUtil

A classe BiaUtil fornece métodos auxiliares para gerenciar headers de requisições e parâmetros de configuração em servidores MCP.

Principais métodos:

• construtor: Recebe uma instância de FastMCP para acessar o contexto das requisições. Armazena a referência do MCP para uso nos métodos.

• __get_from_ssm: Busca um parâmetro no AWS SSM Parameter Store.

• get_header: Extrai e retorna todos os headers customizados da requisição atual. Retorna um objeto Header com os seguintes campos:

  • current_host: Host do ERP no qual o copilot está em execução.
  • user_email: Email do usuário autenticado.
  • jwt_token: Token JWT do usuário -> SankhyaID, SankhyaPass ou Token interno Bia.
  • jsessionid: ID de autenticação do ERP.
  • organization_id: ID da organização da Bia.
  • codparc: Código do parceiro (parceiro Sankhya).
  • iam_user_id: ID do usuário do BIA IAM.
  • gateway_token: Token primário do Sankhya API Gateway.

• get_parameter: Busca o parâmetro informado em duas fontes distintas:

  • Primeiro: Variáveis de ambiente do sistema.
  • Segundo: AWS SSM Parameter Store (se não encontrado nas variáveis de ambiente).
  • Retorna o valor do parâmetro ou None se não encontrado em nenhuma fonte.

Como utilizar

Nesta seção você encontrará uma breve descrição de como utilizar os principais recursos da biblioteca Bia Toolkit.

Criando um MCP Server

Primeiro, instale os pacotes MCP e Bia Toolkit.

pip install mcp biatoolkit

Crie um novo arquivo chamado meu_mcp_server.py com o seguinte conteúdo:

from mcp.server.fastmcp import FastMCP
import json

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

@mcp.tool()
def listar() -> str:
    """Retorna uma lista de exemplo"""
    exemplo = {
        "itens": [
            {"id": 1, "nome": "Item 1"},
            {"id": 2, "nome": "Item 2"},
            {"id": 3, "nome": "Item 3"},
        ]
    }
    return json.dumps(exemplo, indent=4, sort_keys=True)

@mcp.tool()
def adicionar(id: int, nome: str) -> str:
    """Adiciona um item à lista de exemplo"""
    novo_item = {"id": id, "nome": nome}
    return json.dumps(novo_item, indent=4, sort_keys=True)


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

Entendendo o código

• FastMCP: Cria um servidor MCP que pode hospedar suas ferramentas.
• @mcp.tool(): Decorador que transforma suas funções Python em ferramentas MCP.
• Tool: Duas ferramentas simples que demonstram diferentes tipos de operação.

⚠️ IMPORTANTE: Para fazer o deploy no ambiente do Bia Agent Builder, o nome do arquivo, que neste exemplo é meu_mcp_server.py, será utilizado para gerar um pacote Python. Este pacote deverá ser único no Bia Agent Builder. Portanto, no deploy, caso este pacote já exista no Bia Agent Builder, você precisará renomear o seu arquivo.

Iniciando o servidor localmente

Para executar o seu servidor localmente:

python meu_mcp_server.py

Você deverá receber uma mensagem no console semelhante à imagem a seguir:

Como testar o servidor local

Crie um novo arquivo chamado local.py com o seguinte conteúdo:

import asyncio
from biatoolkit.basic_client import BiaClient

MCP_SERVER_URL = "http://0.0.0.0:8000/mcp"
client = BiaClient(MCP_SERVER_URL)

async def list_tools() -> None:
    tools = await client.list_tools()
    for tool in tools.tools:
        print(f"Tool: {tool.name}, Description: {tool.description}")


async def call_tool(tool_name: str, params: dict = None) -> None:
    result = await client.call_tool(tool_name, params)
    print(result.content[0].text)


async def main():
    await list_tools()
    

asyncio.run(main())

Entendendo o código

• BiaClient: É uma classe da biblioteca Bia Toolkit que encapsula um cliente HTTP para comunicação com servidores MCP.
• list_tools: Executa a instrução client.list_tools() para recuperar todas as ferramentas disponíveis no servidor MCP.
• call_tool: Executa a instrução client.call_tool(tool_name, params) para executar uma ferramenta específica do servidor MCP.

Executando o código

⚠️ Certifique-se de que o servidor MCP meu_mcp_server.py ainda esteja em execução.

Execute o arquivo local.py em outro terminal:

python local.py

Você deverá ver a saída no console semelhante a:

Caso deseje testar a execução de uma tool, basta alterar o método main conforme a seguir:

async def main():
    await call_tool("adicionar", {"id": 4, "nome": "Novo item"})

Veja que o método call_tool possui dois parâmetros:

• O primeiro é o nome da tool que queremos executar.
• O segundo é um dicionário com os parâmetros da tool, sendo o segundo (opcional) caso exista.

Ao executar o arquivo local.py, você deverá ver a saída no console semelhante a:

Enviando parâmetros via header

Se seu servidor MCP estiver sendo executado localmente, você conseguirá informar qualquer parâmetro no header da requisição. Entretanto, quando o seu servidor MCP estiver hospedado no ambiente de produção do Bia Agent Builder (AWS Bedrock AgentCore), apenas os parâmetros abaixo podem ser utilizados.

⚠️ QUALQUER OUTRO PARÂMETRO SERÁ IGNORADO PELO SERVIDOR. ⚠️

• X-Amzn-Bedrock-AgentCore-Runtime-Custom-current-host: Host do ERP no qual o copilot está em execução.
• X-Amzn-Bedrock-AgentCore-Runtime-Custom-user-email: Email do usuário autenticado.
• X-Amzn-Bedrock-AgentCore-Runtime-Custom-jwt-token: JWT token do usuário -> SankhyaID, SankhyaPass ou Token interno Bia.
• X-Amzn-Bedrock-AgentCore-Runtime-Custom-jsessionid: ID de autenticação do ERP.
• X-Amzn-Bedrock-AgentCore-Runtime-Custom-organization-id: ID da organização da Bia.
• X-Amzn-Bedrock-AgentCore-Runtime-Custom-codparc: Código do parceiro (parceiro Sankhya).
• X-Amzn-Bedrock-AgentCore-Runtime-Custom-iam-user-id: ID do usuário do BIA IAM.
• X-Amzn-Bedrock-AgentCore-Runtime-Custom-gateway-token: Token primário do Sankhya API Gateway.

import asyncio
from biatoolkit.basic_client import BiaClient

MCP_SERVER_URL = "http://0.0.0.0:8000/mcp"

# Se seu servidor MCP estiver sendo executado **localmente**, você conseguirá informar qualquer parâmetro
# no `header` da requisição. Entretanto, quando o seu servidor MCP estiver hospedado no ambiente de produção
# do Bia Agent Builder (AWS Bedrock AgentCore), apenas os parâmetros abaixo podem ser utilizados.

# Ao utilizar os serviços de interação com a Bia (/agent/stream, /agent/message ou /agent/invoke), 
# os parâmetros abaixo já são automaticamente preenchidos e enviados pelos serviços.
headers = {
    "X-Amzn-Bedrock-AgentCore-Runtime-Custom-current-host": "current-host-123456", # Host do ERP no qual o copilot está em execução.
    "X-Amzn-Bedrock-AgentCore-Runtime-Custom-user-email": "user-email-123456", # Email do usuário autenticado.
    "X-Amzn-Bedrock-AgentCore-Runtime-Custom-jwt-token": "jwt-token-123456", # JWT token do usuário -> SankhyaID, SankhyaPass ou Token interno Bia.
    "X-Amzn-Bedrock-AgentCore-Runtime-Custom-jsessionid": "jsessionid-123456", # ID de autenticação do ERP.
    "X-Amzn-Bedrock-AgentCore-Runtime-Custom-organization-id": "123", # ID da organização da Bia.
    "X-Amzn-Bedrock-AgentCore-Runtime-Custom-codparc": "456", # Código do parceiro (parceiro Sankhya).
    "X-Amzn-Bedrock-AgentCore-Runtime-Custom-iam-user-id": "789", # ID do usuário do BIA IAM.
    "X-Amzn-Bedrock-AgentCore-Runtime-Custom-gateway-token": "gateway-token-123456", # Token primário do Sankhya API Gateway.
    "Content-Type": "application/json"
}

client = BiaClient(MCP_SERVER_URL, headers=headers)

async def list_tools() -> None:
    tools = await client.list_tools()
    for tool in tools.tools:
        print(f"Tool: {tool.name}, Description: {tool.description}")


async def call_tool(tool_name: str, params: dict = None) -> None:
    result = await client.call_tool(tool_name, params)
    print(result.content[0].text)


async def main():
    await call_tool("listar")
    
    
asyncio.run(main())

...existing code...

Para recuperar os parâmetros no MCP Server que foram enviados por meio do header da requisição, basta utilizar a classe BiaUtil conforme a seguir:

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

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

@mcp.tool()
def processar() -> str:
    """Executa o processamento de algo"""
    
    util = BiaUtil(mcp)
    header = util.get_header()

    # Exemplo de uso dos parâmetros do header. Utilize conforme a necessidade 
    # do seu processamento, como autenticação de endpoints, identificação, etc.
    
    print("Valor do parâmetro current_host:", header.current_host)
    print("Valor do parâmetro user_email:", header.user_email)
    print("Valor do parâmetro jwt_token:", header.jwt_token)
    print("Valor do parâmetro jsessionid:", header.jsessionid)
    print("Valor do parâmetro organization_id:", header.organization_id)
    print("Valor do parâmetro codparc:", header.codparc)
    print("Valor do parâmetro iam_user_id:", header.iam_user_id)
    print("Valor do parâmetro gateway_token:", header.gateway_token)
    
    return f"Processamento executado"

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

Recuperando parâmetros do cofre de segredos

Você pode utilizar parâmetros sensíveis de duas formas:

• Usando arquivos .env para execução local.
• Usando o cofre de segredos do Bia Agent Builder para execução em ambiente produtivo.
  • Você pode adicionar parâmetros sensíveis no cofre de segredos do Bia Agent Builder. Para adicionar, alterar e excluir os parâmetros do cofre, utilize as funcionalidades da Plataforma Bia Agent Builder UI.

Para recuperar um valor sensível no seu MCP Server, utilize o método get_parameter(parameter_name: str) da classe BiaUtil.

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

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

@mcp.tool()
def processar() -> str:
    """Executa o processamento de algo"""
    
    util = BiaUtil(mcp)
    valor = util.get_parameter("meu_parametro")

    # Exemplo de uso do parâmetro recuperado. Utilize conforme a necessidade 
    # do seu processamento, como autenticação de endpoints, identificação, etc.

    print("Valor do parâmetro:", valor)
    return f"Processamento executado"

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

O método get_parameter(parameter_name: str) busca o parâmetro informado em duas fontes distintas. Primeiro, o método tenta buscar o parâmetro consultando as variáveis de ambiente do sistema. Caso não exista, o método tenta buscar o parâmetro na AWS SSM Parameter Store.

✅ Isso é vantajoso pois você pode armazenar o parâmetro em:

• Um arquivo .env para testes locais.
• No cofre de segredos do Bia Agent Builder para usar em ambiente produtivo.

Assim, o MESMO CÓDIGO do Servidor MCP pode ser executado em ambos os ambientes.