Metadata-Version: 2.4
Name: biatoolkit
Version: 1.0.3
Summary: Biblioteca para desenvolvedores que utilizam o BiaAgentBuilder
Author: Bia Platform Team
Author-email: data.platform@sankhya.com.br
Classifier: Programming Language :: Python :: 3
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Dynamic: author
Dynamic: author-email
Dynamic: classifier
Dynamic: description
Dynamic: description-content-type
Dynamic: requires-python
Dynamic: summary

# 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**.

### **BasicClient**

A classe `BasicClient` 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.

### **Util**

A classe `Util` 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**.

```bash
pip install mcp biatoolkit
```

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

```python
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:

```bash
python meu_mcp_server.py
```

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

![Figura 01 - Servidor MCP executando localmente](https://iili.io/fSAyNRI.png)

#### **Como testar o servidor local**

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

```python
import asyncio
from biatoolkit.basic_client import BasicClient

MCP_SERVER_URL = "http://0.0.0.0:8000/mcp"
client = BasicClient(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**

- **BasicClient**: É 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:

```bash
python local.py
```

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

![Figura 02 - Cliente MCP listando tools do servidor](https://iili.io/fS7lELN.png)

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

```python
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: 

![Figura 03 - Cliente MCP executando uma tool do servidor](https://iili.io/fS7l1Xp.png)

### **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.

```python
import asyncio
from biatoolkit.basic_client import BasicClient

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 = BasicClient(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())
```

#### **Entendendo o código**

- **headers**: Veja que a variável `headers` é definida com a lista de parâmetros válidos e depois utilizada em `client = BasicClient(MCP_SERVER_URL, headers=headers)`.

⚠️ IMPORTANTE: Ao utilizar os serviços de interação com a Bia (**/agent/stream**, **/agent/message** ou **/agent/invoke**),  os parâmetros já são automaticamente preenchidos e enviados pelos serviços.

### **Recuperando os parâmetros no MCP Server enviados via header**

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

```python
from mcp.server.fastmcp import FastMCP
from biatoolkit.util import Util

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

@mcp.tool()
def processar() -> str:
    """Executa o processamento de algo"""
    
    util = Util(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 `Util`.

```python
from mcp.server.fastmcp import FastMCP
from biatoolkit.util import Util

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

@mcp.tool()
def processar() -> str:
    """Executa o processamento de algo"""
    
    util = Util(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.
