Skip to content

A
aisbf
Commit b6bbf540 authored by Your Name's avatar Your Name
Browse files
Options

feat: Complete kiro-gateway integration with full feature parity 

- Integrated complete kiro-gateway conversion pipeline (1,522 lines)
- Added multi-source authentication (Kiro IDE, kiro-cli, env vars)
- Implemented full OpenAI <-> Kiro format conversion
- Added support for tools/function calling
- Added support for images/multimodal content
- Implemented message merging, validation, and role normalization
- Added KiroAuthManager for automatic token refresh
- Created comprehensive conversion modules:
  - aisbf/kiro_converters.py (core conversion logic)
  - aisbf/kiro_converters_openai.py (OpenAI adapter)
  - aisbf/kiro_models.py (data models)
  - aisbf/kiro_auth.py (authentication)
  - aisbf/kiro_utils.py (utilities)
- Updated KiroProviderHandler with full conversion pipeline
- Added kiro_config support to ProviderConfig
- Updated providers.json with clean Kiro examples
- Added comprehensive documentation (KIRO_INTEGRATION.md)
- Implemented model name prefixing across all providers (provider_id/model)

No external kiro-gateway server needed - all functionality built-in.
parent f8c57389
Show whitespace changes
Inline Side-by-side
Showing with 3600 additions and 4 deletions
AI.PROMPT
View file @ b6bbf540
......@@ -331,6 +331,94 @@ When making changes:
2. Add to `PROVIDER_HANDLERS` dictionary
3. Add provider configuration to `config/providers.json`
### Kiro Gateway Integration
**Overview:**
Kiro Gateway is a third-party proxy gateway that provides OpenAI and Anthropic-compatible APIs for Kiro (Amazon Q Developer / AWS CodeWhisperer). It's located in the `vendor/kiro-gateway` directory and has been integrated as a provider type in AISBF.
**What is Kiro Gateway:**
- Proxy gateway for accessing Claude models through Kiro IDE/CLI credentials
- Provides OpenAI-compatible endpoints (`/v1/chat/completions`, `/v1/models`)
- Supports both Kiro IDE and kiro-cli authentication methods
- Offers access to Claude models (Sonnet 4.5, Haiku 4.5, Opus 4.5, etc.)
- Includes features like extended thinking, tool calling, and streaming
**Integration Architecture:**
- `KiroProviderHandler` class in `aisbf/providers.py` - treats kiro-gateway as an OpenAI-compatible endpoint
- Uses the OpenAI SDK to communicate with kiro-gateway
- Supports all standard AISBF features: streaming, tools, rate limiting, error tracking
**Configuration:**
In `config/providers.json`:
```json
{
"kiro": {
"id": "kiro",
"name": "Kiro Gateway (Amazon Q Developer)",
"endpoint": "http://localhost:8000/v1",
"type": "kiro",
"api_key_required": true,
"rate_limit": 0,
"models": [
{
"name": "claude-sonnet-4-5",
"rate_limit": 0,
"max_request_tokens": 200000,
"context_size": 200000
}
]
}
}
```
In `config/rotations.json`:
```json
{
"kiro-claude": {
"model_name": "kiro-claude",
"providers": [
{
"provider_id": "kiro",
"api_key": "YOUR_KIRO_GATEWAY_API_KEY",
"models": [
{
"name": "claude-sonnet-4-5",
"weight": 3,
"rate_limit": 0
}
]
}
]
}
}
```
**Setup Requirements:**
1. Kiro Gateway must be running (typically on `http://localhost:8000`)
2. Kiro Gateway must be configured with valid Kiro credentials (IDE or CLI)
3. Set `PROXY_API_KEY` in kiro-gateway's `.env` file
4. Use that same API key in AISBF's provider configuration
**Available Models:**
- `claude-sonnet-4-5` - Enhanced model, balanced performance
- `claude-haiku-4-5` - Fast model for quick responses
- `claude-opus-4-5` - Top-tier model (may require paid tier)
- `claude-sonnet-4` - Previous generation model
- `auto` - Automatic model selection
**Usage:**
Once configured, kiro-gateway can be used like any other provider in AISBF:
- Direct provider access: `/api/kiro/chat/completions`
- Rotation access: `/api/kiro-claude/chat/completions`
- Model listing: `/api/kiro/models`
**Benefits:**
- Access Claude models through Kiro credentials without direct Anthropic API access
- Leverage Kiro's free tier or paid plans
- Use Claude models in AISBF rotations alongside other providers
- Automatic failover and load balancing with other providers
### Modifying Configuration
1. Edit files in `~/.aisbf/` for user-specific changes
2. Edit files in installed location for system-wide defaults
......
KIRO_GATEWAY.md 0 → 100644
View file @ b6bbf540
# Kiro Gateway Integration Guide
This guide explains how to use [kiro-gateway](vendor/kiro-gateway) with AISBF to access Claude models through Kiro (Amazon Q Developer / AWS CodeWhisperer) credentials.
## Table of Contents
- [Overview](#overview)
- [What is Kiro Gateway?](#what-is-kiro-gateway)
- [Prerequisites](#prerequisites)
- [Setup Instructions](#setup-instructions)
- [Step 1: Configure Kiro Gateway](#step-1-configure-kiro-gateway)
- [Step 2: Start Kiro Gateway](#step-2-start-kiro-gateway)
- [Step 3: Configure AISBF](#step-3-configure-aisbf)
- [Usage Examples](#usage-examples)
- [Available Models](#available-models)
- [Troubleshooting](#troubleshooting)
- [Architecture](#architecture)
## Overview
Kiro Gateway is a proxy gateway that provides OpenAI and Anthropic-compatible APIs for accessing Claude models through Kiro credentials. By integrating it with AISBF, you can:
- Access Claude models using Kiro IDE or kiro-cli credentials
- Use Claude models in AISBF rotations alongside other providers
- Benefit from automatic failover and load balancing
- Leverage Kiro's free tier or paid plans without direct Anthropic API access
## What is Kiro Gateway?
Kiro Gateway is a standalone FastAPI application located in [`vendor/kiro-gateway`](vendor/kiro-gateway) that:
- Proxies requests to Kiro's backend API
- Provides OpenAI-compatible endpoints (`/v1/chat/completions`, `/v1/models`)
- Provides Anthropic-compatible endpoints (`/v1/messages`)
- Supports both Kiro IDE and kiro-cli authentication methods
- Includes features like extended thinking, tool calling, and streaming
## Prerequisites
Before setting up kiro-gateway with AISBF, ensure you have:
1. **Kiro Credentials**: Either Kiro IDE or kiro-cli configured and authenticated
2. **Python 3.8+**: Required for running kiro-gateway
3. **AISBF Installed**: AISBF should be installed and configured
4. **Network Access**: Both services should be able to communicate (typically on localhost)
## Setup Instructions
### Step 1: Configure Kiro Gateway
1. Navigate to the kiro-gateway directory:
```bash
cd vendor/kiro-gateway
```
2. Install dependencies:
```bash
pip install -r requirements.txt
```
3. Create a `.env` file with your configuration:
```bash
# Authentication method (choose one)
KIRO_AUTH_METHOD=ide # or 'cli'
# For IDE authentication
KIRO_IDE_CONFIG_PATH=/path/to/your/.kiro/config.json
# For CLI authentication
KIRO_CLI_PATH=/path/to/kiro-cli
# API Key for securing the proxy
PROXY_API_KEY=your-secure-api-key-here
# Optional: Server configuration
HOST=0.0.0.0
PORT=8000
LOG_LEVEL=INFO
```
4. **Important**: Choose your authentication method:
**Option A: Kiro IDE Authentication**
- Set `KIRO_AUTH_METHOD=ide`
- Set `KIRO_IDE_CONFIG_PATH` to your Kiro IDE config location
- Default location: `~/.kiro/config.json`
**Option B: kiro-cli Authentication**
- Set `KIRO_AUTH_METHOD=cli`
- Set `KIRO_CLI_PATH` to your kiro-cli executable
- Ensure kiro-cli is authenticated: `kiro-cli auth login`
### Step 2: Start Kiro Gateway
Start the kiro-gateway server:
```bash
cd vendor/kiro-gateway
python main.py
```
You should see output indicating the server is running:
```
INFO: Started server process
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000
```
**Tip**: Run kiro-gateway in a separate terminal or as a background service to keep it running while using AISBF.
### Step 3: Configure AISBF
1. **Add Kiro Provider to `~/.aisbf/providers.json`**:
```json
{
"kiro": {
"id": "kiro",
"name": "Kiro Gateway (Amazon Q Developer)",
"endpoint": "http://localhost:8000/v1",
"type": "kiro",
"api_key_required": true,
"rate_limit": 0,
"models": [
{
"name": "claude-sonnet-4-5",
"rate_limit": 0,
"max_request_tokens": 200000,
"context_size": 200000
},
{
"name": "claude-haiku-4-5",
"rate_limit": 0,
"max_request_tokens": 200000,
"context_size": 200000
},
{
"name": "claude-opus-4-5",
"rate_limit": 0,
"max_request_tokens": 200000,
"context_size": 200000
},
{
"name": "claude-sonnet-4",
"rate_limit": 0,
"max_request_tokens": 200000,
"context_size": 200000
},
{
"name": "auto",
"rate_limit": 0,
"max_request_tokens": 200000,
"context_size": 200000
}
]
}
}
```
**Important**: Replace `YOUR_KIRO_GATEWAY_API_KEY` with the `PROXY_API_KEY` you set in kiro-gateway's `.env` file.
2. **Add Kiro Rotation to `~/.aisbf/rotations.json`** (optional):
```json
{
"kiro-claude": {
"model_name": "kiro-claude",
"providers": [
{
"provider_id": "kiro",
"api_key": "YOUR_KIRO_GATEWAY_API_KEY",
"models": [
{
"name": "claude-sonnet-4-5",
"weight": 3,
"rate_limit": 0
},
{
"name": "claude-haiku-4-5",
"weight": 1,
"rate_limit": 0
}
]
}
]
}
}
```
3. **Restart AISBF** to apply the configuration changes:
```bash
# If running as a service
sudo systemctl restart aisbf
# If running manually
./aisbf.sh
```
## Usage Examples
### Direct Provider Access
Access kiro-gateway directly through AISBF:
```bash
curl -X POST http://localhost:5000/api/kiro/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer YOUR_AISBF_API_KEY" \
-d '{
"model": "claude-sonnet-4-5",
"messages": [
{"role": "user", "content": "Hello, how are you?"}
]
}'
```
### Rotation Access
Use kiro-gateway through a rotation (with automatic model selection):
```bash
curl -X POST http://localhost:5000/api/kiro-claude/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer YOUR_AISBF_API_KEY" \
-d '{
"messages": [
{"role": "user", "content": "Explain quantum computing"}
]
}'
```
### Streaming Responses
Enable streaming for real-time responses:
```bash
curl -X POST http://localhost:5000/api/kiro/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer YOUR_AISBF_API_KEY" \
-d '{
"model": "claude-sonnet-4-5",
"messages": [
{"role": "user", "content": "Write a short story"}
],
"stream": true
}'
```
### Tool Calling
Use Claude's tool calling capabilities:
```bash
curl -X POST http://localhost:5000/api/kiro/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer YOUR_AISBF_API_KEY" \
-d '{
"model": "claude-sonnet-4-5",
"messages": [
{"role": "user", "content": "What is the weather in San Francisco?"}
],
"tools": [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the current weather",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string"}
}
}
}
}
]
}'
```
### List Available Models
Query available models through kiro-gateway:
```bash
curl http://localhost:5000/api/kiro/models \
-H "Authorization: Bearer YOUR_AISBF_API_KEY"
```
## Available Models
Kiro Gateway provides access to the following Claude models:
| Model ID | Description | Context Size | Best For |
|----------|-------------|--------------|----------|
| `claude-sonnet-4-5` | Enhanced Sonnet model | 200K tokens | Balanced performance and quality |
| `claude-haiku-4-5` | Fast Haiku model | 200K tokens | Quick responses, lower cost |
| `claude-opus-4-5` | Top-tier Opus model | 200K tokens | Complex tasks (may require paid tier) |
| `claude-sonnet-4` | Previous generation Sonnet | 200K tokens | Stable, proven performance |
| `auto` | Automatic model selection | Varies | Let Kiro choose the best model |
**Note**: Model availability depends on your Kiro subscription tier. Some models may require a paid plan.
## Troubleshooting
### Kiro Gateway Not Starting
**Problem**: Kiro Gateway fails to start or crashes immediately.
**Solutions**:
1. Check that all dependencies are installed: `pip install -r requirements.txt`
2. Verify your `.env` file has correct configuration
3. Ensure Kiro credentials are valid and authenticated
4. Check logs for specific error messages
### Authentication Errors
**Problem**: Getting 401 Unauthorized errors from kiro-gateway.
**Solutions**:
1. Verify `PROXY_API_KEY` in kiro-gateway's `.env` matches the API key in AISBF's configuration
2. For IDE auth: Check that `KIRO_IDE_CONFIG_PATH` points to a valid config file
3. For CLI auth: Run `kiro-cli auth status` to verify authentication
4. Try re-authenticating: `kiro-cli auth login`
### Connection Refused
**Problem**: AISBF cannot connect to kiro-gateway.
**Solutions**:
1. Verify kiro-gateway is running: `curl http://localhost:8000/health`
2. Check that the endpoint in AISBF's `providers.json` matches kiro-gateway's address
3. Ensure no firewall is blocking the connection
4. Verify the port (default 8000) is not in use by another service
### Model Not Available
**Problem**: Requested model returns an error or is not available.
**Solutions**:
1. Check your Kiro subscription tier - some models require paid plans
2. Use `auto` model to let Kiro select an available model
3. Try a different model (e.g., `claude-haiku-4-5` instead of `claude-opus-4-5`)
4. Check kiro-gateway logs for specific error messages
### Rate Limiting
**Problem**: Requests are being rate limited.
**Solutions**:
1. Check your Kiro account's rate limits
2. Adjust `rate_limit` values in AISBF's configuration
3. Use rotations to distribute load across multiple providers
4. Consider upgrading your Kiro subscription for higher limits
### Streaming Not Working
**Problem**: Streaming responses are not working correctly.
**Solutions**:
1. Ensure `stream: true` is set in the request
2. Check that your client supports Server-Sent Events (SSE)
3. Verify no proxy or middleware is buffering the response
4. Check kiro-gateway logs for streaming-related errors
## Architecture
### Integration Overview
```
┌─────────────┐ ┌─────────────┐ ┌──────────────┐
│ Client │ ──────> │ AISBF │ ──────> │ Kiro Gateway │
│ │ │ Proxy │ │ │
└─────────────┘ └─────────────┘ └──────────────┘
│ │
│ │
v v
┌─────────────┐ ┌──────────────┐
│ Rotations │ │ Kiro API │
│ Failover │ │ (Claude) │
└─────────────┘ └──────────────┘
```
### Request Flow
1. **Client Request**: Client sends request to AISBF endpoint
2. **AISBF Processing**: AISBF validates, routes, and applies rotation logic
3. **Provider Selection**: AISBF selects kiro provider based on configuration
4. **Kiro Gateway**: Request is forwarded to kiro-gateway
5. **Authentication**: Kiro Gateway authenticates using IDE or CLI credentials
6. **Kiro API**: Request is sent to Kiro's backend API
7. **Response**: Response flows back through the chain to the client
### Key Components
- **[`KiroProviderHandler`](aisbf/providers.py)**: Handler class in AISBF that manages kiro-gateway communication
- **OpenAI SDK**: Used to communicate with kiro-gateway's OpenAI-compatible endpoints
- **Kiro Gateway**: Standalone FastAPI proxy in [`vendor/kiro-gateway`](vendor/kiro-gateway)
- **Kiro Credentials**: IDE or CLI authentication for accessing Kiro API
### Benefits of This Architecture
1. **Clean Separation**: Kiro Gateway runs independently, no code duplication
2. **Easy Maintenance**: Update kiro-gateway without modifying AISBF
3. **Flexibility**: Use kiro-gateway with other tools or directly
4. **Standard Interface**: OpenAI-compatible API works with existing tools
5. **Rotation Support**: Combine with other providers for failover and load balancing
## Additional Resources
- [AISBF Documentation](DOCUMENTATION.md)
- [Kiro Gateway README](vendor/kiro-gateway/README.md)
- [Kiro Gateway Architecture](vendor/kiro-gateway/ARCHITECTURE.md)
- [AISBF AI.PROMPT](AI.PROMPT) - Contains integration details and configuration examples
## Support
For issues related to:
- **AISBF**: Check [DOCUMENTATION.md](DOCUMENTATION.md) and [DEBUG_GUIDE.md](DEBUG_GUIDE.md)
- **Kiro Gateway**: Check [`vendor/kiro-gateway/README.md`](vendor/kiro-gateway/README.md)
- **Kiro Credentials**: Refer to Amazon Q Developer / AWS CodeWhisperer documentation
---
**Last Updated**: 2026-03-21
KIRO_INTEGRATION.md 0 → 100644
View file @ b6bbf540
# Kiro Integration for AISBF
## Overview
AISBF now includes **direct integration** with Kiro (Amazon Q Developer), allowing you to use Claude models through your Kiro IDE or kiro-cli credentials without running a separate kiro-gateway server.
This integration was built by analyzing and incorporating the core functionality from the [kiro-gateway](https://github.com/jwadow/kiro-gateway) project directly into AISBF.
## Features
- **Direct API Integration**: Makes API calls directly to Amazon Q Developer's API
- **Multiple Credential Sources**: Supports Kiro IDE, kiro-cli, and environment variables
- **Automatic Token Refresh**: Handles token expiration and refresh automatically
- **No External Dependencies**: No need to run kiro-gateway as a separate service
- **Multiple Authentication Types**: Supports both Kiro Desktop Auth and AWS SSO OIDC
## Architecture
The integration consists of three main components:
1. **`aisbf/kiro_auth.py`**: Authentication manager that handles:
- Loading credentials from multiple sources
- Token refresh for both Kiro Desktop Auth and AWS SSO OIDC
- Automatic token lifecycle management
2. **`aisbf/kiro_utils.py`**: Utility functions for:
- Machine fingerprint generation
- Model name normalization
- Request/response format conversion
3. **`aisbf/providers.py`**: KiroProviderHandler that:
- Uses KiroAuthManager for authentication
- Makes direct HTTP requests to Kiro's API
- Converts between OpenAI format and Kiro format
## Configuration
### Method 1: Using Kiro IDE Credentials (JSON File)
If you have Kiro IDE (VS Code extension) installed, AISBF can use its credentials directly.
**Location**: `~/.config/Code/User/globalStorage/amazon.q/credentials.json`
**Configuration in `~/.aisbf/providers.json`**:
```json
{
"providers": {
"kiro": {
"id": "kiro",
"name": "Kiro (Amazon Q Developer)",
"endpoint": "https://q.us-east-1.amazonaws.com",
"type": "kiro",
"api_key_required": false,
"rate_limit": 0,
"kiro_config": {
"creds_file": "~/.config/Code/User/globalStorage/amazon.q/credentials.json",
"region": "us-east-1"
}
}
}
}
```
### Method 2: Using kiro-cli Credentials (SQLite Database)
If you have kiro-cli installed, AISBF can use its credentials from the SQLite database.
**Location**: `~/.local/share/kiro-cli/data.sqlite3`
**Configuration in `~/.aisbf/providers.json`**:
```json
{
"providers": {
"kiro-cli": {
"id": "kiro-cli",
"name": "Kiro CLI (Amazon Q Developer)",
"endpoint": "https://q.us-east-1.amazonaws.com",
"type": "kiro",
"api_key_required": false,
"rate_limit": 0,
"kiro_config": {
"sqlite_db": "~/.local/share/kiro-cli/data.sqlite3",
"region": "us-east-1"
}
}
}
}
```
### Method 3: Using Environment Variables
You can also provide credentials directly via environment variables or configuration.
**Configuration in `~/.aisbf/providers.json`**:
```json
{
"providers": {
"kiro": {
"id": "kiro",
"name": "Kiro (Amazon Q Developer)",
"endpoint": "https://q.us-east-1.amazonaws.com",
"type": "kiro",
"api_key_required": false,
"rate_limit": 0,
"kiro_config": {
"refresh_token": "your-refresh-token-here",
"profile_arn": "arn:aws:codewhisperer:us-east-1:...",
"client_id": "your-client-id",
"client_secret": "your-client-secret",
"region": "us-east-1"
}
}
}
}
```
## Available Models
The following Claude models are available through Kiro:
- `anthropic.claude-3-5-sonnet-20241022-v2:0` - Claude 3.5 Sonnet v2 (latest)
- `anthropic.claude-3-5-haiku-20241022-v1:0` - Claude 3.5 Haiku
- `anthropic.claude-3-5-sonnet-20240620-v1:0` - Claude 3.5 Sonnet v1
- `anthropic.claude-sonnet-3-5-v2` - Claude 3.5 Sonnet v2 (alias)
## Usage Examples
### Using Kiro Provider Directly
```bash
# Using Kiro provider with a specific model
curl -X POST http://localhost:8080/v1/chat/completions \
-H "Content-Type: application/json" \
-d '{
"provider_id": "kiro",
"model": "anthropic.claude-3-5-sonnet-20241022-v2:0",
"messages": [
{"role": "user", "content": "Hello, how are you?"}
]
}'
```
### Using Kiro in a Rotation
Add Kiro to your rotation configuration in `~/.aisbf/rotations.json`:
```json
{
"rotations": {
"kiro-claude": {
"providers": [
{
"provider_id": "kiro",
"model": "anthropic.claude-3-5-sonnet-20241022-v2:0",
"weight": 1
},
{
"provider_id": "kiro-cli",
"model": "anthropic.claude-3-5-sonnet-20241022-v2:0",
"weight": 1
}
],
"notifyerrors": true
}
}
}
```
Then use the rotation:
```bash
curl -X POST http://localhost:8080/v1/chat/completions \
-H "Content-Type: application/json" \
-d '{
"rotation_id": "kiro-claude",
"messages": [
{"role": "user", "content": "Hello, how are you?"}
]
}'
```
## Authentication Types
### Kiro Desktop Auth
Used by Kiro IDE (VS Code extension). Credentials are stored in JSON format.
- **Token URL**: `https://prod.{region}.auth.desktop.kiro.dev/refreshToken`
- **Method**: POST with JSON body containing refresh token
- **Response**: Access token with expiration time
### AWS SSO OIDC
Used by kiro-cli. Credentials are stored in SQLite database.
- **Token URL**: `https://oidc.{region}.amazonaws.com/token`
- **Method**: POST with form data (grant_type, client_id, client_secret, refresh_token)
- **Response**: Access token with expiration time
## Regions
Kiro supports multiple AWS regions:
- `us-east-1` (default)
- `eu-central-1`
- `ap-southeast-1`
- `us-west-2`
The API endpoint is always `https://q.{region}.amazonaws.com`, regardless of the region.
## Troubleshooting
### "Kiro authentication not configured"
**Cause**: No valid credentials found in any of the configured sources.
**Solution**:
1. Verify that Kiro IDE or kiro-cli is installed and authenticated
2. Check that the credential file/database path is correct
3. Ensure the credentials haven't expired
### "Profile ARN not available"
**Cause**: The profile ARN couldn't be loaded from credentials.
**Solution**:
1. Re-authenticate with Kiro IDE or kiro-cli
2. Verify that your AWS account has access to Amazon Q Developer
3. Check that the credentials file contains a valid profile ARN
### Token Refresh Failures
**Cause**: Refresh token has expired or is invalid.
**Solution**:
1. Re-authenticate with Kiro IDE or kiro-cli
2. Check that your AWS credentials are still valid
3. Verify network connectivity to AWS endpoints
### "Improperly formed request"
**Cause**: The request format doesn't match Kiro API requirements.
**Solution**:
1. Check that you're using a valid model ID
2. Ensure messages are properly formatted
3. Review the logs for specific error details
## Logging
Enable debug logging to see detailed information about Kiro API calls:
```bash
export AISBF_DEBUG=true
python -m aisbf.main
```
This will show:
- Authentication attempts and token refresh
- API request/response details
- Credential loading process
- Error details
## Comparison with kiro-gateway
### Before (External kiro-gateway)
```
Client → AISBF → kiro-gateway (separate process) → Kiro API
```
**Drawbacks**:
- Need to run kiro-gateway as a separate service
- Additional network hop
- More complex deployment
### After (Direct Integration)
```
Client → AISBF → Kiro API
```
**Benefits**:
- No external dependencies
- Simpler deployment
- Lower latency
- Unified configuration
## Credits
This integration was built by analyzing and incorporating functionality from:
- [kiro-gateway](https://github.com/jwadow/kiro-gateway) by Jwadow
The core authentication and API interaction logic was adapted from kiro-gateway's implementation.
## License
The Kiro integration code in AISBF is licensed under the GNU General Public License v3.0, consistent with AISBF's license.
The original kiro-gateway project is licensed under the GNU Affero General Public License v3.0.
aisbf/config.py
View file @ b6bbf540
......@@ -52,6 +52,7 @@ class ProviderConfig(BaseModel):
rate_limit: float = 0.0
api_key: Optional[str] = None # Optional API key in provider config
models: Optional[List[ProviderModelConfig]] = None # Optional list of models with their configs
kiro_config: Optional[Dict] = None # Optional Kiro-specific configuration (credentials, region, etc.)
class RotationConfig(BaseModel):
providers: List[Dict]
......
aisbf/kiro_auth.py 0 → 100644
View file @ b6bbf540
"""
Kiro Authentication Module for AISBF
Adapted from kiro-gateway's auth.py
"""
import asyncio
import json
import sqlite3
from datetime import datetime, timezone, timedelta
from enum import Enum
from pathlib import Path
from typing import Optional, Dict, Any
import httpx
import hashlib
import uuid
import socket
import getpass
import logging
logger = logging.getLogger(__name__)
class AuthType(Enum):
"""Authentication type enum"""
KIRO_DESKTOP = "kiro_desktop"
AWS_SSO_OIDC = "aws_sso_oidc"
class KiroAuthManager:
"""
Kiro Authentication Manager
Handles authentication for Kiro API (Amazon Q Developer/CodeWhisperer)
Supports both Kiro Desktop Auth and AWS SSO OIDC (kiro-cli)
"""
def __init__(self,
refresh_token: Optional[str] = None,
profile_arn: Optional[str] = None,
region: str = "us-east-1",
creds_file: Optional[str] = None,
sqlite_db: Optional[str] = None,
client_id: Optional[str] = None,
client_secret: Optional[str] = None):
"""
Initialize Kiro Authentication Manager
Args:
refresh_token: Refresh token for Kiro Desktop Auth
profile_arn: AWS CodeWhisperer profile ARN
region: AWS region (default: us-east-1)
creds_file: Path to JSON credentials file
sqlite_db: Path to kiro-cli SQLite database
client_id: OAuth client ID (for AWS SSO OIDC)
client_secret: OAuth client secret (for AWS SSO OIDC)
"""
self.refresh_token = refresh_token
self.profile_arn = profile_arn
self.region = region
self.creds_file = creds_file
self.sqlite_db = sqlite_db
self.client_id = client_id
self.client_secret = client_secret
self._access_token = None
self._refresh_token = refresh_token
self._expires_at = None
self._auth_type = AuthType.KIRO_DESKTOP
self._lock = asyncio.Lock()
# SQLite token keys to search for
self.SQLITE_TOKEN_KEYS = [
"kirocli:social:token",
"kirocli:odic:token",
"codewhisperer:odic:token"
]
self.SQLITE_REG_KEYS = [
"kirocli:odic:device-registration",
"codewhisperer:odic:device-registration"
]
# Load credentials from file or SQLite if provided
if sqlite_db:
self._load_from_sqlite()
elif creds_file:
self._load_from_creds_file()
# Determine auth type
self._detect_auth_type()
def _detect_auth_type(self):
"""Detect authentication type based on available credentials"""
if self.client_id and self.client_secret:
self._auth_type = AuthType.AWS_SSO_OIDC
logger.info("Detected AWS SSO OIDC authentication")
else:
self._auth_type = AuthType.KIRO_DESKTOP
logger.info("Detected Kiro Desktop authentication")
def _load_from_sqlite(self):
"""Load credentials from SQLite database (kiro-cli)"""
if not self.sqlite_db:
return
try:
conn = sqlite3.connect(self.sqlite_db)
cursor = conn.cursor()
# Try to find token in SQLite
token_data = None
token_key = None
for key in self.SQLITE_TOKEN_KEYS:
cursor.execute("SELECT value FROM auth_kv WHERE key = ?", (key,))
row = cursor.fetchone()
if row:
token_data = json.loads(row[0])
token_key = key
break
if token_data:
self._access_token = token_data.get('access_token')
self._refresh_token = token_data.get('refresh_token')
self._expires_at = datetime.fromisoformat(
token_data.get('expires_at', '1970-01-01T00:00:00Z')
)
logger.info(f"Loaded credentials from SQLite key: {token_key}")
# Try to get device registration for AWS SSO OIDC
for reg_key in self.SQLITE_REG_KEYS:
cursor.execute("SELECT value FROM auth_kv WHERE key = ?", (reg_key,))
row = cursor.fetchone()
if row:
reg_data = json.loads(row[0])
self.client_id = reg_data.get('clientId')
self.client_secret = reg_data.get('clientSecret')
break
except Exception as e:
logger.error(f"Failed to load from SQLite: {e}")
finally:
conn.close()
def _load_from_creds_file(self):
"""Load credentials from JSON file"""
if not self.creds_file:
return
try:
with open(self.creds_file, 'r') as f:
data = json.load(f)
self.refresh_token = data.get('refreshToken', self.refresh_token)
self._access_token = data.get('accessToken')
self.profile_arn = data.get('profileArn', self.profile_arn)
if 'expiresAt' in data:
self._expires_at = datetime.fromisoformat(
data['expiresAt'].replace('Z', '+00:00')
)
except Exception as e:
logger.error(f"Failed to load credentials file: {e}")
def _get_machine_fingerprint(self):
"""Generate machine fingerprint for User-Agent"""
try:
hostname = socket.gethostname()
username = getpass.getuser()
unique_string = f"{hostname}-{username}-kiro-gateway"
return hashlib.sha256(unique_string.encode()).hexdigest()
except:
return hashlib.sha256(b"default-machine-fingerprint").hexdigest()
async def get_access_token(self) -> str:
"""Get a valid access token, refreshing if necessary"""
async with self._lock:
if self._access_token and self._expires_at and self._expires_at > datetime.now(timezone.utc):
return self._access_token
# Need to refresh token
if self._auth_type == AuthType.AWS_SSO_OIDC:
await self._refresh_aws_sso_token()
else:
await self._refresh_kiro_desktop_token()
return self._access_token
async def _refresh_kiro_desktop_token(self):
"""Refresh token using Kiro Desktop Auth"""
if not self.refresh_token:
raise ValueError("No refresh token available")
url = f"https://prod.{self.region}.auth.desktop.kiro.dev/refreshToken"
headers = {
"Content-Type": "application/json",
"User-Agent": f"KiroIDE-0.7.45-{self._get_machine_fingerprint()}"
}
payload = {"refreshToken": self.refresh_token}
async with httpx.AsyncClient() as client:
response = await client.post(url, json=payload, headers=headers)
response.raise_for_status()
data = response.json()
self._access_token = data['accessToken']
if 'refreshToken' in data:
self.refresh_token = data['refreshToken']
# Calculate expiration (1 hour default)
self._expires_at = datetime.now(timezone.utc) + timedelta(seconds=3600)
# Save if we have a credentials file
if self.creds_file:
self._save_credentials()
async def _refresh_aws_sso_token(self):
"""Refresh token using AWS SSO OIDC"""
if not all([self.refresh_token, self.client_id, self.client_secret]):
raise ValueError("Missing credentials for AWS SSO OIDC")
url = f"https://oidc.{self.region}.amazonaws.com/token"
payload = {
"grant_type": "refresh_token",
"client_id": self.client_id,
"client_secret": self.client_secret,
"refresh_token": self.refresh_token
}
async with httpx.AsyncClient() as client:
response = await client.post(url, data=payload)
response.raise_for_status()
data = response.json()
self._access_token = data['access_token']
if 'refresh_token' in data:
self.refresh_token = data['refresh_token']
expires_in = data.get('expires_in', 3600)
self._expires_at = datetime.now(timezone.utc) + timedelta(seconds=expires_in)
def _save_credentials(self):
"""Save updated credentials to file"""
if not self.creds_file:
return
try:
with open(self.creds_file, 'r') as f:
data = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
data = {}
data.update({
'accessToken': self._access_token,
'refreshToken': self.refresh_token,
'expiresAt': self._expires_at.isoformat() if self._expires_at else None,
'profileArn': self.profile_arn,
'region': self.region
})
with open(self.creds_file, 'w') as f:
json.dump(data, f, indent=2)
def get_auth_headers(self, token: str) -> dict:
"""Get headers for Kiro API requests"""
fingerprint = self._get_machine_fingerprint()
return {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"User-Agent": f"aws-sdk-js/1.0.27 KiroIDE-0.7.45-{fingerprint}",
"x-amz-user-agent": f"aws-sdk-js/1.0.27 KiroIDE-0.7.45-{fingerprint}",
"x-amz-codewhisperer-optout": "true",
"x-amzn-kiro-agent-mode": "vibe",
"amz-sdk-invocation-id": str(uuid.uuid4()),
"amz-sdk-request": "attempt=1; max=3"
}
def _get_machine_fingerprint(self):
"""Get machine fingerprint for User-Agent"""
try:
import socket
import getpass
hostname = socket.gethostname()
username = getpass.getuser()
unique_string = f"{hostname}-{username}-kiro-gateway"
return hashlib.sha256(unique_string.encode()).hexdigest()
except:
return hashlib.sha256(b"default-machine-fingerprint").hexdigest()
\ No newline at end of file
aisbf/kiro_converters.py 0 → 100644
View file @ b6bbf540
# -*- coding: utf-8 -*-
# Kiro Gateway
# https://github.com/jwadow/kiro-gateway
# Copyright (C) 2025 Jwadow
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""
Core converters for transforming API formats to Kiro format.
This module contains shared logic used by both OpenAI and Anthropic converters:
- Text content extraction from various formats
- Message merging and processing
- Kiro payload building
- Tool processing and sanitization
The core layer provides a unified interface that API-specific adapters use
to convert their formats to Kiro API format.
"""
import json
import logging
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple
# Use standard Python logging instead of loguru
logger = logging.getLogger(__name__)
# Configuration constants (from kiro-gateway config)
TOOL_DESCRIPTION_MAX_LENGTH = 1000 # Max length for tool descriptions
FAKE_REASONING_ENABLED = False # Disable fake reasoning by default
FAKE_REASONING_MAX_TOKENS = 32000 # Max tokens for thinking
TRUNCATION_RECOVERY = False # Disable truncation recovery by default
# ==================================================================================================
# Data Classes for Unified Message Format
# ==================================================================================================
@dataclass
class UnifiedMessage:
"""
Unified message format used internally by converters.
This format is API-agnostic and can be created from both OpenAI and Anthropic formats.
Serves as the canonical representation for all message data before conversion to Kiro API.
Attributes:
role: Message role (user, assistant, system)
content: Text content or list of content blocks
tool_calls: List of tool calls (for assistant messages)
tool_results: List of tool results (for user messages with tool responses)
images: List of images in unified format (for multimodal user messages)
Format: [{"media_type": "image/jpeg", "data": "base64..."}]
"""
role: str
content: Any = ""
tool_calls: Optional[List[Dict[str, Any]]] = None
tool_results: Optional[List[Dict[str, Any]]] = None
images: Optional[List[Dict[str, Any]]] = None
@dataclass
class UnifiedTool:
"""
Unified tool format used internally by converters.
Attributes:
name: Tool name
description: Tool description
input_schema: JSON Schema for tool parameters
"""
name: str
description: Optional[str] = None
input_schema: Optional[Dict[str, Any]] = None
@dataclass
class KiroPayloadResult:
"""
Result of building Kiro payload.
Attributes:
payload: The complete Kiro API payload
tool_documentation: Documentation for tools with long descriptions (to add to system prompt)
"""
payload: Dict[str, Any]
tool_documentation: str = ""
# ==================================================================================================
# Text Content Extraction
# ==================================================================================================
def extract_text_content(content: Any) -> str:
"""
Extracts text content from various formats.
Supports multiple content formats used by different APIs:
- String: "Hello, world!"
- List of content blocks: [{"type": "text", "text": "Hello"}]
- None: empty message
Args:
content: Content in any supported format
Returns:
Extracted text or empty string
Example:
>>> extract_text_content("Hello")
'Hello'
>>> extract_text_content([{"type": "text", "text": "World"}])
'World'
>>> extract_text_content(None)
''
"""
if content is None:
return ""
if isinstance(content, str):
return content
if isinstance(content, list):
text_parts = []
for item in content:
if isinstance(item, dict):
# Skip image blocks - they're handled separately
if item.get("type") in ("image", "image_url"):
continue
if item.get("type") == "text":
text_parts.append(item.get("text", ""))
elif "text" in item:
text_parts.append(item["text"])
elif hasattr(item, "text"):
# Handle Pydantic models like TextContentBlock
text_parts.append(getattr(item, "text", ""))
elif isinstance(item, str):
text_parts.append(item)
return "".join(text_parts)
return str(content)
def extract_images_from_content(content: Any) -> List[Dict[str, Any]]:
"""
Extracts images from message content in unified format.
Supports multiple image formats used by different APIs:
OpenAI format (image_url with data URL):
{"type": "image_url", "image_url": {"url": "data:image/jpeg;base64,/9j/..."}}
Anthropic format (image with source):
{"type": "image", "source": {"type": "base64", "media_type": "image/jpeg", "data": "/9j/..."}}
Args:
content: Content in any supported format (usually a list of content blocks)
Returns:
List of images in unified format: [{"media_type": "image/jpeg", "data": "base64..."}]
Empty list if no images found or content is not a list.
Example:
>>> extract_images_from_content([{"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": "abc123"}}])
[{'media_type': 'image/png', 'data': 'abc123'}]
"""
images: List[Dict[str, Any]] = []
if not isinstance(content, list):
return images
for item in content:
# Handle both dict and Pydantic model objects
if isinstance(item, dict):
item_type = item.get("type")
elif hasattr(item, "type"):
item_type = item.type
else:
continue
# OpenAI format: {"type": "image_url", "image_url": {"url": "data:image/jpeg;base64,..."}}
if item_type == "image_url":
if isinstance(item, dict):
image_url_obj = item.get("image_url", {})
else:
image_url_obj = getattr(item, "image_url", {})
if isinstance(image_url_obj, dict):
url = image_url_obj.get("url", "")
elif hasattr(image_url_obj, "url"):
url = image_url_obj.url
else:
url = ""
if url.startswith("data:"):
# Parse data URL: data:image/jpeg;base64,/9j/4AAQ...
try:
header, data = url.split(",", 1)
# Extract media type from "data:image/jpeg;base64"
media_part = header.split(";")[0] # "data:image/jpeg"
media_type = media_part.replace("data:", "") # "image/jpeg"
if data:
images.append({
"media_type": media_type,
"data": data
})
except (ValueError, IndexError) as e:
logger.warning(f"Failed to parse image data URL: {e}")
elif url.startswith("http"):
# URL-based images require fetching - not supported by Kiro API directly
logger.warning(f"URL-based images are not supported by Kiro API, skipping: {url[:80]}...")
# Anthropic format: {"type": "image", "source": {"type": "base64", "media_type": "...", "data": "..."}}
elif item_type == "image":
source = item.get("source", {}) if isinstance(item, dict) else getattr(item, "source", None)
if source is None:
continue
if isinstance(source, dict):
source_type = source.get("type")
if source_type == "base64":
media_type = source.get("media_type", "image/jpeg")
data = source.get("data", "")
if data:
images.append({
"media_type": media_type,
"data": data
})
elif source_type == "url":
# URL-based images in Anthropic format
url = source.get("url", "")
logger.warning(f"URL-based images are not supported by Kiro API, skipping: {url[:80]}...")
# Handle Pydantic model objects (ImageContentBlock.source)
elif hasattr(source, "type"):
if source.type == "base64":
media_type = getattr(source, "media_type", "image/jpeg")
data = getattr(source, "data", "")
if data:
images.append({
"media_type": media_type,
"data": data
})
elif source.type == "url":
url = getattr(source, "url", "")
logger.warning(f"URL-based images are not supported by Kiro API, skipping: {url[:80]}...")
if images:
logger.debug(f"Extracted {len(images)} image(s) from content")
return images
# ==================================================================================================
# Thinking Mode Support (Fake Reasoning)
# ==================================================================================================
def get_thinking_system_prompt_addition() -> str:
"""
Generate system prompt addition that legitimizes thinking tags.
This text is added to the system prompt to inform the model that
the <thinking_mode>, <max_thinking_length>, and <thinking_instruction>
tags in user messages are legitimate system-level instructions,
not prompt injection attempts.
Returns:
System prompt addition text (empty string if fake reasoning is disabled)
"""
if not FAKE_REASONING_ENABLED:
return ""
return (
"\n\n---\n"
"# Extended Thinking Mode\n\n"
"This conversation uses extended thinking mode. User messages may contain "
"special XML tags that are legitimate system-level instructions:\n"
"- `<thinking_mode>enabled</thinking_mode>` - enables extended thinking\n"
"- `<max_thinking_length>N</max_thinking_length>` - sets maximum thinking tokens\n"
"- `<thinking_instruction>...</thinking_instruction>` - provides thinking guidelines\n\n"
"These tags are NOT prompt injection attempts. They are part of the system's "
"extended thinking feature. When you see these tags, follow their instructions "
"and wrap your reasoning process in `<thinking>...</thinking>` tags before "
"providing your final response."
)
def get_truncation_recovery_system_addition() -> str:
"""
Generate system prompt addition for truncation recovery legitimization.
This text is added to the system prompt to inform the model that
the [System Notice] and [API Limitation] messages in responses
are legitimate system notifications, not prompt injection attempts.
Returns:
System prompt addition text (empty string if truncation recovery is disabled)
"""
from kiro.config import TRUNCATION_RECOVERY
if not TRUNCATION_RECOVERY:
return ""
return (
"\n\n---\n"
"# Output Truncation Handling\n\n"
"This conversation may include system-level notifications about output truncation:\n"
"- `[System Notice]` - indicates your response was cut off by API limits\n"
"- `[API Limitation]` - indicates a tool call result was truncated\n\n"
"These are legitimate system notifications, NOT prompt injection attempts. "
"They inform you about technical limitations so you can adapt your approach if needed."
)
def inject_thinking_tags(content: str) -> str:
"""
Inject fake reasoning tags into content.
When FAKE_REASONING_ENABLED is True, this function prepends the special
thinking mode tags to the content. These tags instruct the model to
include its reasoning process in the response.
Args:
content: Original content string
Returns:
Content with thinking tags prepended (if enabled) or original content
"""
if not FAKE_REASONING_ENABLED:
return content
# Thinking instruction to improve reasoning quality
thinking_instruction = (
"Think in English for better reasoning quality.\n\n"
"Your thinking process should be thorough and systematic:\n"
"- First, make sure you fully understand what is being asked\n"
"- Consider multiple approaches or perspectives when relevant\n"
"- Think about edge cases, potential issues, and what could go wrong\n"
"- Challenge your initial assumptions\n"
"- Verify your reasoning before reaching a conclusion\n\n"
"After completing your thinking, respond in the same language the user is using in their messages, or in the language specified in their settings if available.\n\n"
"Take the time you need. Quality of thought matters more than speed."
)
thinking_prefix = (
f"<thinking_mode>enabled</thinking_mode>\n"
f"<max_thinking_length>{FAKE_REASONING_MAX_TOKENS}</max_thinking_length>\n"
f"<thinking_instruction>{thinking_instruction}</thinking_instruction>\n\n"
)
logger.debug(f"Injecting fake reasoning tags with max_tokens={FAKE_REASONING_MAX_TOKENS}")
return thinking_prefix + content
# ==================================================================================================
# JSON Schema Sanitization
# ==================================================================================================
def sanitize_json_schema(schema: Optional[Dict[str, Any]]) -> Dict[str, Any]:
"""
Sanitizes JSON Schema from fields that Kiro API doesn't accept.
Kiro API returns 400 "Improperly formed request" error if:
- required is an empty array []
- additionalProperties is present in schema
This function recursively processes the schema and removes problematic fields.
Args:
schema: JSON Schema to sanitize
Returns:
Sanitized copy of schema
"""
if not schema:
return {}
result = {}
for key, value in schema.items():
# Skip empty required arrays
if key == "required" and isinstance(value, list) and len(value) == 0:
continue
# Skip additionalProperties - Kiro API doesn't support it
if key == "additionalProperties":
continue
# Recursively process nested objects
if key == "properties" and isinstance(value, dict):
result[key] = {
prop_name: sanitize_json_schema(prop_value) if isinstance(prop_value, dict) else prop_value
for prop_name, prop_value in value.items()
}
elif isinstance(value, dict):
result[key] = sanitize_json_schema(value)
elif isinstance(value, list):
# Process lists (e.g., anyOf, oneOf)
result[key] = [
sanitize_json_schema(item) if isinstance(item, dict) else item
for item in value
]
else:
result[key] = value
return result
# ==================================================================================================
# Tool Processing
# ==================================================================================================
def process_tools_with_long_descriptions(
tools: Optional[List[UnifiedTool]]
) -> Tuple[Optional[List[UnifiedTool]], str]:
"""
Processes tools with long descriptions.
Kiro API has a limit on description length in toolSpecification.
If description exceeds the limit, full description is moved to system prompt,
and a reference to documentation remains in the tool.
Args:
tools: List of tools in unified format
Returns:
Tuple of:
- List of tools with processed descriptions (or None if tools is empty)
- String with documentation to add to system prompt (empty if all descriptions are short)
"""
if not tools:
return None, ""
# If limit is disabled (0), return tools unchanged
if TOOL_DESCRIPTION_MAX_LENGTH <= 0:
return tools, ""
tool_documentation_parts = []
processed_tools = []
for tool in tools:
description = tool.description or ""
if len(description) <= TOOL_DESCRIPTION_MAX_LENGTH:
# Description is short - leave as is
processed_tools.append(tool)
else:
# Description is too long - move to system prompt
logger.debug(
f"Tool '{tool.name}' has long description ({len(description)} chars > {TOOL_DESCRIPTION_MAX_LENGTH}), "
f"moving to system prompt"
)
# Create documentation for system prompt
tool_documentation_parts.append(f"## Tool: {tool.name}\n\n{description}")
# Create copy of tool with reference description
reference_description = f"[Full documentation in system prompt under '## Tool: {tool.name}']"
processed_tool = UnifiedTool(
name=tool.name,
description=reference_description,
input_schema=tool.input_schema
)
processed_tools.append(processed_tool)
# Form final documentation
tool_documentation = ""
if tool_documentation_parts:
tool_documentation = (
"\n\n---\n"
"# Tool Documentation\n"
"The following tools have detailed documentation that couldn't fit in the tool definition.\n\n"
+ "\n\n---\n\n".join(tool_documentation_parts)
)
return processed_tools if processed_tools else None, tool_documentation
def validate_tool_names(tools: Optional[List[UnifiedTool]]) -> None:
"""
Validates tool names against Kiro API 64-character limit.
Logs WARNING for each problematic tool and raises ValueError
with complete list of violations.
Args:
tools: List of tools to validate
Raises:
ValueError: If any tool name exceeds 64 characters
Example:
>>> validate_tool_names([UnifiedTool(name="short_name", description="test")])
# No error
>>> validate_tool_names([UnifiedTool(name="a" * 70, description="test")])
# Raises ValueError with detailed message
"""
if not tools:
return
problematic_tools = []
for tool in tools:
if len(tool.name) > 64:
problematic_tools.append((tool.name, len(tool.name)))
if problematic_tools:
# Build detailed error message for client (no logging here - routes will log)
tool_list = "\n".join([
f" - '{name}' ({length} characters)"
for name, length in problematic_tools
])
raise ValueError(
f"Tool name(s) exceed Kiro API limit of 64 characters:\n"
f"{tool_list}\n\n"
f"Solution: Use shorter tool names (max 64 characters).\n"
f"Example: 'get_user_data' instead of 'get_authenticated_user_profile_data_with_extended_information_about_it'"
)
def convert_tools_to_kiro_format(tools: Optional[List[UnifiedTool]]) -> List[Dict[str, Any]]:
"""
Converts unified tools to Kiro API format.
Args:
tools: List of tools in unified format
Returns:
List of tools in Kiro toolSpecification format
"""
if not tools:
return []
kiro_tools = []
for tool in tools:
# Sanitize parameters from fields that Kiro API doesn't accept
sanitized_params = sanitize_json_schema(tool.input_schema)
# Kiro API requires non-empty description
description = tool.description
if not description or not description.strip():
description = f"Tool: {tool.name}"
logger.debug(f"Tool '{tool.name}' has empty description, using placeholder")
kiro_tools.append({
"toolSpecification": {
"name": tool.name,
"description": description,
"inputSchema": {"json": sanitized_params}
}
})
return kiro_tools
# ==================================================================================================
# Image Conversion to Kiro Format
# ==================================================================================================
def convert_images_to_kiro_format(images: Optional[List[Dict[str, Any]]]) -> List[Dict[str, Any]]:
"""
Converts unified images to Kiro API format.
Unified format: [{"media_type": "image/jpeg", "data": "base64..."}]
Kiro format: [{"format": "jpeg", "source": {"bytes": "base64..."}}]
IMPORTANT: Images must be placed directly in userInputMessage.images,
NOT in userInputMessageContext.images. This matches the native Kiro IDE format.
Also handles the case where data contains a full data URL (data:image/jpeg;base64,...)
by stripping the prefix and extracting pure base64.
Args:
images: List of images in unified format
Returns:
List of images in Kiro format, ready for userInputMessage.images
Example:
>>> convert_images_to_kiro_format([{"media_type": "image/png", "data": "abc123"}])
[{'format': 'png', 'source': {'bytes': 'abc123'}}]
"""
if not images:
return []
kiro_images = []
for img in images:
media_type = img.get("media_type", "image/jpeg")
data = img.get("data", "")
if not data:
logger.warning("Skipping image with empty data")
continue
# Strip data URL prefix if present (some clients send "data:image/jpeg;base64,..." in data field)
# Kiro API expects pure base64 without the prefix
if data.startswith("data:"):
try:
header, actual_data = data.split(",", 1)
# Extract media type from header if present
media_part = header.split(";")[0] # "data:image/jpeg"
extracted_media_type = media_part.replace("data:", "")
if extracted_media_type:
media_type = extracted_media_type
data = actual_data
logger.debug(f"Stripped data URL prefix, extracted media_type: {media_type}")
except (ValueError, IndexError) as e:
logger.warning(f"Failed to parse data URL prefix: {e}")
# Extract format from media_type: "image/jpeg" -> "jpeg"
format_str = media_type.split("/")[-1] if "/" in media_type else media_type
kiro_images.append({
"format": format_str,
"source": {
"bytes": data
}
})
if kiro_images:
logger.debug(f"Converted {len(kiro_images)} image(s) to Kiro format")
return kiro_images
# ==================================================================================================
# Tool Results and Tool Uses Extraction
# ==================================================================================================
def convert_tool_results_to_kiro_format(tool_results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Converts unified tool results to Kiro API format.
Unified format: {"type": "tool_result", "tool_use_id": "...", "content": "..."}
Kiro format: {"content": [{"text": "..."}], "status": "success", "toolUseId": "..."}
Args:
tool_results: List of tool results in unified format
Returns:
List of tool results in Kiro format
"""
kiro_results = []
for tr in tool_results:
content = tr.get("content", "")
if isinstance(content, str):
content_text = content
else:
content_text = extract_text_content(content)
# Ensure content is not empty - Kiro API requires non-empty content
if not content_text:
content_text = "(empty result)"
kiro_results.append({
"content": [{"text": content_text}],
"status": "success",
"toolUseId": tr.get("tool_use_id", "")
})
return kiro_results
def extract_tool_results_from_content(content: Any) -> List[Dict[str, Any]]:
"""
Extracts tool results from message content.
Looks for content blocks with type="tool_result" and converts them
to Kiro API format.
Args:
content: Message content (can be a list of content blocks)
Returns:
List of tool results in Kiro format
"""
tool_results = []
if isinstance(content, list):
for item in content:
if isinstance(item, dict) and item.get("type") == "tool_result":
tool_results.append({
"content": [{"text": extract_text_content(item.get("content", "")) or "(empty result)"}],
"status": "success",
"toolUseId": item.get("tool_use_id", "")
})
return tool_results
def extract_tool_uses_from_message(
content: Any,
tool_calls: Optional[List[Dict[str, Any]]] = None
) -> List[Dict[str, Any]]:
"""
Extracts tool uses from assistant message.
Looks for tool calls in both:
- tool_calls field (OpenAI format)
- content blocks with type="tool_use" (Anthropic format)
Args:
content: Message content
tool_calls: List of tool calls (OpenAI format)
Returns:
List of tool uses in Kiro format
"""
tool_uses = []
# From tool_calls field (OpenAI format or unified format from Anthropic)
if tool_calls:
for tc in tool_calls:
if isinstance(tc, dict):
func = tc.get("function", {})
arguments = func.get("arguments", "{}")
# Handle both string (OpenAI) and dict (Anthropic unified) formats
if isinstance(arguments, str):
input_data = json.loads(arguments) if arguments else {}
else:
input_data = arguments if arguments else {}
tool_uses.append({
"name": func.get("name", ""),
"input": input_data,
"toolUseId": tc.get("id", "")
})
# From content blocks (Anthropic format)
if isinstance(content, list):
for item in content:
if isinstance(item, dict) and item.get("type") == "tool_use":
tool_uses.append({
"name": item.get("name", ""),
"input": item.get("input", {}),
"toolUseId": item.get("id", "")
})
return tool_uses
# ==================================================================================================
# Tool Content to Text Conversion (for stripping when no tools defined)
# ==================================================================================================
def tool_calls_to_text(tool_calls: List[Dict[str, Any]]) -> str:
"""
Converts tool_calls to human-readable text representation.
This is used when stripping tool content from messages (when no tools are defined).
Instead of losing the context, we convert tool calls to text so the model
can still understand what happened in the conversation.
Args:
tool_calls: List of tool calls in unified format
Returns:
Text representation of tool calls
Example:
>>> tool_calls_to_text([{"id": "call_123", "function": {"name": "bash", "arguments": '{"command": "ls"}'}}])
'[Tool: bash] (call_123)\\n{"command": "ls"}'
"""
if not tool_calls:
return ""
parts = []
for tc in tool_calls:
func = tc.get("function", {})
name = func.get("name", "unknown")
arguments = func.get("arguments", "{}")
tool_id = tc.get("id", "")
# Format: [Tool: name] (id)\narguments
if tool_id:
parts.append(f"[Tool: {name} ({tool_id})]\n{arguments}")
else:
parts.append(f"[Tool: {name}]\n{arguments}")
return "\n\n".join(parts)
def tool_results_to_text(tool_results: List[Dict[str, Any]]) -> str:
"""
Converts tool_results to human-readable text representation.
This is used when stripping tool content from messages (when no tools are defined).
Instead of losing the context, we convert tool results to text so the model
can still understand what happened in the conversation.
Args:
tool_results: List of tool results in unified format
Returns:
Text representation of tool results
Example:
>>> tool_results_to_text([{"tool_use_id": "call_123", "content": "file1.txt\\nfile2.txt"}])
'[Tool Result] (call_123)\\nfile1.txt\\nfile2.txt'
"""
if not tool_results:
return ""
parts = []
for tr in tool_results:
content = tr.get("content", "")
tool_use_id = tr.get("tool_use_id", "")
if isinstance(content, str):
content_text = content
else:
content_text = extract_text_content(content)
# Use placeholder if content is empty
if not content_text:
content_text = "(empty result)"
# Format: [Tool Result] (id)\ncontent
if tool_use_id:
parts.append(f"[Tool Result ({tool_use_id})]\n{content_text}")
else:
parts.append(f"[Tool Result]\n{content_text}")
return "\n\n".join(parts)
# ==================================================================================================
# Message Merging
# ==================================================================================================
def strip_all_tool_content(messages: List[UnifiedMessage]) -> Tuple[List[UnifiedMessage], bool]:
"""
Strips ALL tool-related content from messages, converting it to text representation.
This is used when no tools are defined in the request. Kiro API rejects
requests that have toolResults but no tools defined.
Instead of simply removing tool content, this function converts tool_calls
and tool_results to human-readable text, preserving the context for
summarization and other use cases.
Args:
messages: List of messages in unified format
Returns:
Tuple of:
- List of messages with tool content converted to text
- Boolean indicating whether any tool content was converted
"""
if not messages:
return [], False
result = []
total_tool_calls_stripped = 0
total_tool_results_stripped = 0
for msg in messages:
# Check if this message has any tool content
has_tool_calls = bool(msg.tool_calls)
has_tool_results = bool(msg.tool_results)
if has_tool_calls or has_tool_results:
if has_tool_calls:
total_tool_calls_stripped += len(msg.tool_calls)
if has_tool_results:
total_tool_results_stripped += len(msg.tool_results)
# Start with existing text content
existing_content = extract_text_content(msg.content)
content_parts = []
if existing_content:
content_parts.append(existing_content)
# Convert tool_calls to text (for assistant messages)
if has_tool_calls:
tool_text = tool_calls_to_text(msg.tool_calls)
if tool_text:
content_parts.append(tool_text)
# Convert tool_results to text (for user messages)
if has_tool_results:
result_text = tool_results_to_text(msg.tool_results)
if result_text:
content_parts.append(result_text)
# Join all parts with double newline
content = "\n\n".join(content_parts) if content_parts else "(empty)"
# Create a copy of the message without tool content but with text representation
# IMPORTANT: Preserve images from the original message (e.g., screenshots from MCP tools)
cleaned_msg = UnifiedMessage(
role=msg.role,
content=content,
tool_calls=None,
tool_results=None,
images=msg.images
)
result.append(cleaned_msg)
else:
result.append(msg)
had_tool_content = total_tool_calls_stripped > 0 or total_tool_results_stripped > 0
# Log summary once (DEBUG level - this is normal for clients like Cline/Roo/Cursor)
if had_tool_content:
logger.debug(
f"Converted tool content to text (no tools defined): "
f"{total_tool_calls_stripped} tool_calls, {total_tool_results_stripped} tool_results"
)
return result, had_tool_content
def ensure_assistant_before_tool_results(messages: List[UnifiedMessage]) -> Tuple[List[UnifiedMessage], bool]:
"""
Ensures that messages with tool_results have a preceding assistant message with tool_calls.
Kiro API requires that when toolResults are present, there must be a preceding
assistantResponseMessage with toolUses. Some clients (like Cline/Roo/Cursor) may send
truncated conversations where the assistant message is missing.
Since we don't know the original tool name and arguments when the assistant message
is missing, we cannot create a valid synthetic assistant message. Instead, we convert
the tool_results to text representation and append to the message content, preserving
the context for the model while avoiding Kiro API rejection.
Args:
messages: List of messages in unified format
Returns:
Tuple of:
- List of messages with orphaned tool_results converted to text
- Boolean indicating whether any tool_results were converted (used to skip thinking tag injection)
"""
if not messages:
return [], False
result = []
converted_any_tool_results = False
for msg in messages:
# Check if this message has tool_results
if msg.tool_results:
# Check if the previous message is an assistant with tool_calls
has_preceding_assistant = (
result and
result[-1].role == "assistant" and
result[-1].tool_calls
)
if not has_preceding_assistant:
# We cannot create a valid synthetic assistant message because we don't know
# the original tool name and arguments. Kiro API validates tool names.
# Convert tool_results to text to preserve context for the model.
logger.debug(
f"Converting {len(msg.tool_results)} orphaned tool_results to text "
f"(no preceding assistant message with tool_calls). "
f"Tool IDs: {[tr.get('tool_use_id', 'unknown') for tr in msg.tool_results]}"
)
# Convert tool_results to text representation
tool_results_text = tool_results_to_text(msg.tool_results)
# Append to existing content
original_content = extract_text_content(msg.content) or ""
if original_content and tool_results_text:
new_content = f"{original_content}\n\n{tool_results_text}"
elif tool_results_text:
new_content = tool_results_text
else:
new_content = original_content
# Create a copy of the message with tool_results converted to text
cleaned_msg = UnifiedMessage(
role=msg.role,
content=new_content,
tool_calls=msg.tool_calls,
tool_results=None, # Remove orphaned tool_results (now in text)
images=msg.images
)
result.append(cleaned_msg)
converted_any_tool_results = True
continue
result.append(msg)
return result, converted_any_tool_results
def merge_adjacent_messages(messages: List[UnifiedMessage]) -> List[UnifiedMessage]:
"""
Merges adjacent messages with the same role.
Kiro API does not accept multiple consecutive messages from the same role.
This function merges such messages into one.
Args:
messages: List of messages in unified format
Returns:
List of messages with merged adjacent messages
"""
if not messages:
return []
merged = []
# Statistics for summary logging
merge_counts = {"user": 0, "assistant": 0}
total_tool_calls_merged = 0
total_tool_results_merged = 0
for msg in messages:
if not merged:
merged.append(msg)
continue
last = merged[-1]
if msg.role == last.role:
# Merge content
if isinstance(last.content, list) and isinstance(msg.content, list):
last.content = last.content + msg.content
elif isinstance(last.content, list):
last.content = last.content + [{"type": "text", "text": extract_text_content(msg.content)}]
elif isinstance(msg.content, list):
last.content = [{"type": "text", "text": extract_text_content(last.content)}] + msg.content
else:
last_text = extract_text_content(last.content)
current_text = extract_text_content(msg.content)
last.content = f"{last_text}\n{current_text}"
# Merge tool_calls for assistant messages
if msg.role == "assistant" and msg.tool_calls:
if last.tool_calls is None:
last.tool_calls = []
last.tool_calls = list(last.tool_calls) + list(msg.tool_calls)
total_tool_calls_merged += len(msg.tool_calls)
# Merge tool_results for user messages
if msg.role == "user" and msg.tool_results:
if last.tool_results is None:
last.tool_results = []
last.tool_results = list(last.tool_results) + list(msg.tool_results)
total_tool_results_merged += len(msg.tool_results)
# Count merges by role
if msg.role in merge_counts:
merge_counts[msg.role] += 1
else:
merged.append(msg)
# Log summary if any merges occurred
total_merges = sum(merge_counts.values())
if total_merges > 0:
parts = []
for role, count in merge_counts.items():
if count > 0:
parts.append(f"{count} {role}")
merge_summary = ", ".join(parts)
extras = []
if total_tool_calls_merged > 0:
extras.append(f"{total_tool_calls_merged} tool_calls")
if total_tool_results_merged > 0:
extras.append(f"{total_tool_results_merged} tool_results")
if extras:
logger.debug(f"Merged {total_merges} adjacent messages ({merge_summary}), including {', '.join(extras)}")
else:
logger.debug(f"Merged {total_merges} adjacent messages ({merge_summary})")
return merged
def ensure_first_message_is_user(messages: List[UnifiedMessage]) -> List[UnifiedMessage]:
"""
Ensures that the first message in the conversation is from user role.
Kiro API requires conversations to start with a user message. If the first
message is from assistant (or any other non-user role), we prepend a minimal
synthetic user message.
This matches LiteLLM behavior for Anthropic API compatibility and fixes
issue #60 where conversations starting with assistant messages cause
"Improperly formed request" errors.
Args:
messages: List of messages in unified format
Returns:
List of messages with guaranteed user-first order
Example:
>>> messages = [
... UnifiedMessage(role="assistant", content="Hello"),
... UnifiedMessage(role="user", content="Hi")
... ]
>>> result = ensure_first_message_is_user(messages)
>>> result[0].role
'user'
>>> result[0].content
'(empty)'
"""
if not messages:
return messages
if messages[0].role != "user":
logger.debug(
f"First message is '{messages[0].role}', prepending synthetic user message "
f"(Kiro API requires conversations to start with user)"
)
# Create minimal synthetic user message (matches LiteLLM behavior)
# Using "(empty)" as minimal valid content to avoid disrupting conversation context
synthetic_user = UnifiedMessage(
role="user",
content="(empty)"
)
return [synthetic_user] + messages
return messages
def normalize_message_roles(messages: List[UnifiedMessage]) -> List[UnifiedMessage]:
"""
Normalizes unknown message roles to 'user'.
Kiro API only supports 'user' and 'assistant' roles in history.
Any other role (e.g., 'developer', 'system') is converted to 'user'
to maintain compatibility.
This normalization MUST happen before ensure_alternating_roles()
to ensure consecutive messages with unknown roles are properly detected
and synthetic assistant messages are inserted between them.
Args:
messages: List of messages in unified format
Returns:
List of messages with normalized roles
Example:
>>> messages = [
... UnifiedMessage(role="developer", content="Context 1"),
... UnifiedMessage(role="developer", content="Context 2"),
... UnifiedMessage(role="user", content="Question")
... ]
>>> result = normalize_message_roles(messages)
>>> [msg.role for msg in result]
['user', 'user', 'user']
"""
if not messages:
return messages
normalized = []
converted_count = 0
for msg in messages:
if msg.role not in ("user", "assistant"):
logger.debug(f"Normalizing role '{msg.role}' to 'user'")
normalized_msg = UnifiedMessage(
role="user",
content=msg.content,
tool_calls=msg.tool_calls,
tool_results=msg.tool_results,
images=msg.images
)
normalized.append(normalized_msg)
converted_count += 1
else:
normalized.append(msg)
if converted_count > 0:
logger.debug(f"Normalized {converted_count} message(s) with unknown roles to 'user'")
return normalized
def ensure_alternating_roles(messages: List[UnifiedMessage]) -> List[UnifiedMessage]:
"""
Ensures alternating user/assistant roles by inserting synthetic assistant messages.
Kiro API requires alternating userInputMessage and assistantResponseMessage.
When consecutive user messages are detected, synthetic assistant messages
with "(empty)" placeholder are inserted between them to maintain alternation.
This fixes multiple unknown roles (converted to user)
create consecutive userInputMessage entries that violate Kiro API requirements.
Args:
messages: List of messages in unified format
Returns:
List of messages with synthetic assistant messages inserted where needed
Example:
>>> messages = [
... UnifiedMessage(role="user", content="First"),
... UnifiedMessage(role="user", content="Second"),
... UnifiedMessage(role="user", content="Third")
... ]
>>> result = ensure_alternating_roles(messages)
>>> len(result)
5 # 3 user + 2 synthetic assistant
>>> result[1].role
'assistant'
>>> result[1].content
'(empty)'
"""
if not messages or len(messages) < 2:
return messages
result = [messages[0]]
synthetic_count = 0
for msg in messages[1:]:
prev_role = result[-1].role
# If both current and previous are user → insert synthetic assistant
if msg.role == "user" and prev_role == "user":
synthetic_assistant = UnifiedMessage(
role="assistant",
content="(empty)" # Consistent with build_kiro_history() placeholder
)
result.append(synthetic_assistant)
synthetic_count += 1
result.append(msg)
if synthetic_count > 0:
logger.debug(f"Inserted {synthetic_count} synthetic assistant message(s) to ensure alternation")
return result
# ==================================================================================================
# Kiro History Building
# ==================================================================================================
def build_kiro_history(messages: List[UnifiedMessage], model_id: str) -> List[Dict[str, Any]]:
"""
Builds history array for Kiro API from unified messages.
Kiro API expects alternating userInputMessage and assistantResponseMessage.
This function converts unified format to Kiro format.
All messages should have 'user' or 'assistant' roles at this point,
as unknown roles are normalized earlier in the pipeline by normalize_message_roles().
Args:
messages: List of messages in unified format (with normalized roles)
model_id: Internal Kiro model ID
Returns:
List of dictionaries for history field in Kiro API
"""
history = []
for msg in messages:
if msg.role == "user":
content = extract_text_content(msg.content)
# Fallback for empty content - Kiro API requires non-empty content
if not content:
content = "(empty)"
user_input = {
"content": content,
"modelId": model_id,
"origin": "AI_EDITOR",
}
# Process images - extract from message or content
# IMPORTANT: images go directly into userInputMessage, NOT into userInputMessageContext
# This matches the native Kiro IDE format
images = msg.images or extract_images_from_content(msg.content)
if images:
kiro_images = convert_images_to_kiro_format(images)
if kiro_images:
user_input["images"] = kiro_images
# Build userInputMessageContext for tools and toolResults only
user_input_context: Dict[str, Any] = {}
# Process tool_results - convert to Kiro format if present
if msg.tool_results:
kiro_tool_results = convert_tool_results_to_kiro_format(msg.tool_results)
if kiro_tool_results:
user_input_context["toolResults"] = kiro_tool_results
else:
# Try to extract from content (already in Kiro format)
tool_results = extract_tool_results_from_content(msg.content)
if tool_results:
user_input_context["toolResults"] = tool_results
# Add context if not empty (contains toolResults only, not images)
if user_input_context:
user_input["userInputMessageContext"] = user_input_context
history.append({"userInputMessage": user_input})
elif msg.role == "assistant":
content = extract_text_content(msg.content)
# Fallback for empty content - Kiro API requires non-empty content
if not content:
content = "(empty)"
assistant_response = {"content": content}
# Process tool_calls
tool_uses = extract_tool_uses_from_message(msg.content, msg.tool_calls)
if tool_uses:
assistant_response["toolUses"] = tool_uses
history.append({"assistantResponseMessage": assistant_response})
return history
# ==================================================================================================
# Main Payload Building
# ==================================================================================================
def build_kiro_payload(
messages: List[UnifiedMessage],
system_prompt: str,
model_id: str,
tools: Optional[List[UnifiedTool]],
conversation_id: str,
profile_arn: str,
inject_thinking: bool = True
) -> KiroPayloadResult:
"""
Builds complete payload for Kiro API from unified data.
This is the main function that assembles the Kiro API payload from
API-agnostic unified message and tool formats.
Args:
messages: List of messages in unified format (without system messages)
system_prompt: Already extracted system prompt
model_id: Internal Kiro model ID
tools: List of tools in unified format (or None)
conversation_id: Unique conversation ID
profile_arn: AWS CodeWhisperer profile ARN
inject_thinking: Whether to inject thinking tags (default True)
Returns:
KiroPayloadResult with payload and tool documentation
Raises:
ValueError: If there are no messages to send
"""
# Process tools with long descriptions
processed_tools, tool_documentation = process_tools_with_long_descriptions(tools)
# Validate tool names against Kiro API 64-character limit
validate_tool_names(processed_tools)
# Add tool documentation to system prompt if present
full_system_prompt = system_prompt
if tool_documentation:
full_system_prompt = full_system_prompt + tool_documentation if full_system_prompt else tool_documentation.strip()
# Add thinking mode legitimization to system prompt if enabled
thinking_system_addition = get_thinking_system_prompt_addition()
if thinking_system_addition:
full_system_prompt = full_system_prompt + thinking_system_addition if full_system_prompt else thinking_system_addition.strip()
# Add truncation recovery legitimization to system prompt if enabled
truncation_system_addition = get_truncation_recovery_system_addition()
if truncation_system_addition:
full_system_prompt = full_system_prompt + truncation_system_addition if full_system_prompt else truncation_system_addition.strip()
# If no tools are defined, strip ALL tool-related content from messages
# Kiro API rejects requests with toolResults but no tools
if not tools:
messages_without_tools, had_tool_content = strip_all_tool_content(messages)
messages_with_assistants = messages_without_tools
converted_tool_results = had_tool_content
else:
# Ensure assistant messages exist before tool_results (Kiro API requirement)
# Also returns flag if any tool_results were converted (to skip thinking tag injection)
messages_with_assistants, converted_tool_results = ensure_assistant_before_tool_results(messages)
# Merge adjacent messages with the same role
merged_messages = merge_adjacent_messages(messages_with_assistants)
# Ensure first message is from user (Kiro API requirement, fixes issue #60)
merged_messages = ensure_first_message_is_user(merged_messages)
# Normalize unknown roles to 'user' (fixes issue #64)
# This must happen BEFORE ensure_alternating_roles() so that consecutive
# messages with unknown roles (e.g., 'developer') are properly detected
merged_messages = normalize_message_roles(merged_messages)
# Ensure alternating user/assistant roles (fixes issue #64)
# Insert synthetic assistant messages between consecutive user messages
merged_messages = ensure_alternating_roles(merged_messages)
if not merged_messages:
raise ValueError("No messages to send")
# Build history (all messages except the last one)
history_messages = merged_messages[:-1] if len(merged_messages) > 1 else []
# If there's a system prompt, add it to the first user message in history
if full_system_prompt and history_messages:
first_msg = history_messages[0]
if first_msg.role == "user":
original_content = extract_text_content(first_msg.content)
first_msg.content = f"{full_system_prompt}\n\n{original_content}"
history = build_kiro_history(history_messages, model_id)
# Current message (the last one)
current_message = merged_messages[-1]
current_content = extract_text_content(current_message.content)
# If system prompt exists but history is empty - add to current message
if full_system_prompt and not history:
current_content = f"{full_system_prompt}\n\n{current_content}"
# If current message is assistant, need to add it to history
# and create user message "Continue"
if current_message.role == "assistant":
history.append({
"assistantResponseMessage": {
"content": current_content
}
})
current_content = "Continue"
# If content is empty - use "Continue"
if not current_content:
current_content = "Continue"
# Process images in current message - extract from message or content
# IMPORTANT: images go directly into userInputMessage, NOT into userInputMessageContext
# This matches the native Kiro IDE format
images = current_message.images or extract_images_from_content(current_message.content)
kiro_images = None
if images:
kiro_images = convert_images_to_kiro_format(images)
if kiro_images:
logger.debug(f"Added {len(kiro_images)} image(s) to current message")
# Build user_input_context for tools and toolResults only (NOT images)
user_input_context: Dict[str, Any] = {}
# Add tools if present
kiro_tools = convert_tools_to_kiro_format(processed_tools)
if kiro_tools:
user_input_context["tools"] = kiro_tools
# Process tool_results in current message - convert to Kiro format if present
if current_message.tool_results:
# Convert unified format to Kiro format
kiro_tool_results = convert_tool_results_to_kiro_format(current_message.tool_results)
if kiro_tool_results:
user_input_context["toolResults"] = kiro_tool_results
else:
# Try to extract from content (already in Kiro format)
tool_results = extract_tool_results_from_content(current_message.content)
if tool_results:
user_input_context["toolResults"] = tool_results
# Inject thinking tags if enabled (only for the current/last user message)
if inject_thinking and current_message.role == "user":
current_content = inject_thinking_tags(current_content)
# Build userInputMessage
user_input_message = {
"content": current_content,
"modelId": model_id,
"origin": "AI_EDITOR",
}
# Add images directly to userInputMessage (NOT to userInputMessageContext)
if kiro_images:
user_input_message["images"] = kiro_images
# Add user_input_context if present (contains tools and toolResults only)
if user_input_context:
user_input_message["userInputMessageContext"] = user_input_context
# Assemble final payload
payload = {
"conversationState": {
"chatTriggerType": "MANUAL",
"conversationId": conversation_id,
"currentMessage": {
"userInputMessage": user_input_message
}
}
}
# Add history only if not empty
if history:
payload["conversationState"]["history"] = history
# Add profileArn
if profile_arn:
payload["profileArn"] = profile_arn
return KiroPayloadResult(payload=payload, tool_documentation=tool_documentation)
\ No newline at end of file
aisbf/kiro_converters_openai.py 0 → 100644
View file @ b6bbf540
# -*- coding: utf-8 -*-
# Kiro Gateway
# https://github.com/jwadow/kiro-gateway
# Copyright (C) 2025 Jwadow
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""
Converters for transforming OpenAI format to Kiro format.
This module is an adapter layer that converts OpenAI-specific formats
to the unified format used by converters_core.py.
Contains functions for:
- Converting OpenAI messages to unified format
- Converting OpenAI tools to unified format
- Building Kiro payload from OpenAI requests
"""
import logging
from typing import Any, Dict, List, Optional, Tuple
# Use standard Python logging
logger = logging.getLogger(__name__)
# Configuration - hidden models that need normalization
HIDDEN_MODELS = {
"claude-sonnet-4-5": "anthropic.claude-3-5-sonnet-20241022-v2:0",
"claude-haiku-4-5": "anthropic.claude-3-5-haiku-20241022-v1:0",
"claude-opus-4-5": "anthropic.claude-3-5-opus-20250514-v1:0",
"claude-sonnet-4": "anthropic.claude-3-5-sonnet-20240620-v1:0",
}
def get_model_id_for_kiro(model: str, hidden_models: dict) -> str:
"""Normalize model name for Kiro API"""
return hidden_models.get(model, model)
# Import from core - reuse shared logic
from .kiro_converters import (
extract_text_content,
extract_images_from_content,
UnifiedMessage,
UnifiedTool,
build_kiro_payload as core_build_kiro_payload,
)
# ==================================================================================================
# OpenAI-specific Message Processing
# ==================================================================================================
def _extract_tool_results_from_openai(content: Any) -> List[Dict[str, Any]]:
"""
Extracts tool results from OpenAI message content.
Args:
content: Message content (can be a list with tool_result blocks)
Returns:
List of tool results in unified format for UnifiedMessage
"""
tool_results = []
if isinstance(content, list):
for item in content:
if isinstance(item, dict) and item.get("type") == "tool_result":
tool_results.append({
"type": "tool_result",
"tool_use_id": item.get("tool_use_id", ""),
"content": extract_text_content(item.get("content", "")) or "(empty result)"
})
return tool_results
def _extract_images_from_tool_message(content: Any) -> List[Dict[str, Any]]:
"""
Extracts images from OpenAI tool message content.
Tool messages from MCP servers (e.g., browsermcp) can contain images
(screenshots) alongside text. This function extracts those images.
Args:
content: Tool message content (can be string or list of content blocks)
Returns:
List of images in unified format: [{"media_type": "image/jpeg", "data": "base64..."}]
Example:
>>> content = [
... {"type": "text", "text": "Screenshot captured"},
... {"type": "image_url", "image_url": {"url": "data:image/png;base64,..."}}
... ]
>>> images = _extract_images_from_tool_message(content)
>>> len(images)
1
"""
# If content is not a list, no images to extract
if not isinstance(content, list):
return []
# Use core function to extract images from content list
images = extract_images_from_content(content)
if images:
logger.debug(f"Extracted {len(images)} image(s) from tool message content")
return images
def _extract_tool_calls_from_openai(msg: ChatMessage) -> List[Dict[str, Any]]:
"""
Extracts tool calls from OpenAI assistant message.
Args:
msg: OpenAI ChatMessage
Returns:
List of tool calls in unified format
"""
tool_calls = []
if msg.tool_calls:
for tc in msg.tool_calls:
if isinstance(tc, dict):
tool_calls.append({
"id": tc.get("id", ""),
"type": "function",
"function": {
"name": tc.get("function", {}).get("name", ""),
"arguments": tc.get("function", {}).get("arguments", "{}")
}
})
return tool_calls
def convert_openai_messages_to_unified(messages: List[ChatMessage]) -> Tuple[str, List[UnifiedMessage]]:
"""
Converts OpenAI messages to unified format.
Handles:
- System messages (extracted as system prompt)
- Tool messages (converted to user messages with tool_results)
- Tool calls in assistant messages
Args:
messages: List of OpenAI ChatMessage objects
Returns:
Tuple of (system_prompt, unified_messages)
"""
# Extract system prompt
system_prompt = ""
non_system_messages = []
for msg in messages:
if msg.role == "system":
system_prompt += extract_text_content(msg.content) + "\n"
else:
non_system_messages.append(msg)
system_prompt = system_prompt.strip()
# Process tool messages - convert to user messages with tool_results
processed = []
pending_tool_results = []
pending_tool_images = []
total_tool_calls = 0
total_tool_results = 0
total_images = 0
for msg in non_system_messages:
if msg.role == "tool":
# Collect tool results
tool_result = {
"type": "tool_result",
"tool_use_id": msg.tool_call_id or "",
"content": extract_text_content(msg.content) or "(empty result)"
}
pending_tool_results.append(tool_result)
total_tool_results += 1
# Extract images from tool message content (e.g., screenshots from MCP tools)
tool_images = _extract_images_from_tool_message(msg.content)
if tool_images:
pending_tool_images.extend(tool_images)
total_images += len(tool_images)
else:
# If there are accumulated tool results, create user message with them
if pending_tool_results:
unified_msg = UnifiedMessage(
role="user",
content="",
tool_results=pending_tool_results.copy(),
images=pending_tool_images.copy() if pending_tool_images else None
)
processed.append(unified_msg)
pending_tool_results.clear()
pending_tool_images.clear()
# Convert regular message
tool_calls = None
tool_results = None
images = None
if msg.role == "assistant":
tool_calls = _extract_tool_calls_from_openai(msg) or None
if tool_calls:
total_tool_calls += len(tool_calls)
elif msg.role == "user":
tool_results = _extract_tool_results_from_openai(msg.content) or None
if tool_results:
total_tool_results += len(tool_results)
# Extract images from user messages
images = extract_images_from_content(msg.content) or None
if images:
total_images += len(images)
unified_msg = UnifiedMessage(
role=msg.role,
content=extract_text_content(msg.content),
tool_calls=tool_calls,
tool_results=tool_results,
images=images
)
processed.append(unified_msg)
# If tool results remain at the end
if pending_tool_results:
unified_msg = UnifiedMessage(
role="user",
content="",
tool_results=pending_tool_results.copy(),
images=pending_tool_images.copy() if pending_tool_images else None
)
processed.append(unified_msg)
# Log summary if any tool content or images were found
if total_tool_calls > 0 or total_tool_results > 0 or total_images > 0:
logger.debug(
f"Converted {len(messages)} OpenAI messages: "
f"{total_tool_calls} tool_calls, {total_tool_results} tool_results, {total_images} images"
)
return system_prompt, processed
def convert_openai_tools_to_unified(tools: Optional[List[Tool]]) -> Optional[List[UnifiedTool]]:
"""
Converts OpenAI tools to unified format.
Supports two formats:
1. Standard OpenAI format: {"type": "function", "function": {"name": "...", ...}}
2. Flat format (Cursor-style): {"name": "...", "description": "...", "input_schema": {...}}
Args:
tools: List of OpenAI Tool objects
Returns:
List of UnifiedTool objects, or None if no tools
"""
if not tools:
return None
unified_tools = []
for tool in tools:
if tool.type != "function":
continue
# Standard OpenAI format (function field) takes priority
if tool.function is not None:
unified_tools.append(UnifiedTool(
name=tool.function.name,
description=tool.function.description,
input_schema=tool.function.parameters
))
# Flat format compatibility (Cursor-style)
elif tool.name is not None:
unified_tools.append(UnifiedTool(
name=tool.name,
description=tool.description,
input_schema=tool.input_schema
))
# Skip invalid tools
else:
logger.warning(f"Skipping invalid tool: no function or name field found")
continue
return unified_tools if unified_tools else None
# ==================================================================================================
# Main Entry Point
# ==================================================================================================
def build_kiro_payload_from_dict(
model: str,
messages: list,
tools: list = None,
conversation_id: str = None,
profile_arn: str = None
) -> dict:
"""
Builds complete payload for Kiro API from dict-based OpenAI request.
This is a convenience wrapper that converts dicts to dataclasses
and then uses the full conversion pipeline.
Args:
model: Model name
messages: List of message dicts
tools: Optional list of tool dicts
conversation_id: Unique conversation ID
profile_arn: AWS CodeWhisperer profile ARN
Returns:
Payload dictionary for POST request to Kiro API
Raises:
ValueError: If there are no messages to send
"""
from .kiro_models import create_chat_completion_request
# Convert dicts to dataclasses
request_data = create_chat_completion_request(
model=model,
messages=messages,
tools=tools
)
# Convert messages to unified format
system_prompt, unified_messages = convert_openai_messages_to_unified(request_data.messages)
# Convert tools to unified format
unified_tools = convert_openai_tools_to_unified(request_data.tools)
# Get model ID for Kiro API (normalizes + resolves hidden models)
model_id = get_model_id_for_kiro(request_data.model, HIDDEN_MODELS)
logger.debug(
f"Converting OpenAI request: model={request_data.model} -> {model_id}, "
f"messages={len(unified_messages)}, tools={len(unified_tools) if unified_tools else 0}, "
f"system_prompt_length={len(system_prompt)}"
)
# Use core function to build payload
result = core_build_kiro_payload(
messages=unified_messages,
system_prompt=system_prompt,
model_id=model_id,
tools=unified_tools,
conversation_id=conversation_id or "",
profile_arn=profile_arn or "",
inject_thinking=True
)
return result.payload
\ No newline at end of file
aisbf/kiro_models.py 0 → 100644
View file @ b6bbf540
"""
Data models for Kiro integration.
Simple dataclasses to represent OpenAI-style requests for the Kiro converters.
"""
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Union
@dataclass
class FunctionDefinition:
"""Function definition for tools"""
name: str
description: Optional[str] = None
parameters: Optional[Dict[str, Any]] = None
@dataclass
class Tool:
"""Tool definition"""
type: str = "function"
function: Optional[FunctionDefinition] = None
# Flat format support (Cursor-style)
name: Optional[str] = None
description: Optional[str] = None
input_schema: Optional[Dict[str, Any]] = None
@dataclass
class ChatMessage:
"""Chat message"""
role: str
content: Optional[Union[str, List[Dict[str, Any]]]] = None
tool_calls: Optional[List[Dict[str, Any]]] = None
tool_call_id: Optional[str] = None
name: Optional[str] = None
@dataclass
class ChatCompletionRequest:
"""Chat completion request"""
model: str
messages: List[ChatMessage]
tools: Optional[List[Tool]] = None
temperature: Optional[float] = 1.0
max_tokens: Optional[int] = None
stream: Optional[bool] = False
def dict_to_chat_message(msg_dict: Dict[str, Any]) -> ChatMessage:
"""Convert dict to ChatMessage"""
return ChatMessage(
role=msg_dict.get("role", "user"),
content=msg_dict.get("content"),
tool_calls=msg_dict.get("tool_calls"),
tool_call_id=msg_dict.get("tool_call_id"),
name=msg_dict.get("name")
)
def dict_to_tool(tool_dict: Dict[str, Any]) -> Tool:
"""Convert dict to Tool"""
tool_type = tool_dict.get("type", "function")
# Standard OpenAI format
if "function" in tool_dict:
func_dict = tool_dict["function"]
function = FunctionDefinition(
name=func_dict.get("name", ""),
description=func_dict.get("description"),
parameters=func_dict.get("parameters")
)
return Tool(type=tool_type, function=function)
# Flat format (Cursor-style)
return Tool(
type=tool_type,
name=tool_dict.get("name"),
description=tool_dict.get("description"),
input_schema=tool_dict.get("input_schema") or tool_dict.get("parameters")
)
def create_chat_completion_request(
model: str,
messages: List[Dict[str, Any]],
tools: Optional[List[Dict[str, Any]]] = None,
temperature: Optional[float] = 1.0,
max_tokens: Optional[int] = None,
stream: Optional[bool] = False
) -> ChatCompletionRequest:
"""Create ChatCompletionRequest from dicts"""
chat_messages = [dict_to_chat_message(msg) for msg in messages]
chat_tools = [dict_to_tool(tool) for tool in tools] if tools else None
return ChatCompletionRequest(
model=model,
messages=chat_messages,
tools=chat_tools,
temperature=temperature,
max_tokens=max_tokens,
stream=stream
)
aisbf/kiro_utils.py 0 → 100644
View file @ b6bbf540
"""
Kiro Utilities for AISBF
Adapted from kiro-gateway utils.py
"""
import hashlib
import uuid
import socket
import getpass
import json
from typing import Dict, Any, List, Optional
import hashlib
import uuid as uuid_lib
def get_machine_fingerprint() -> str:
"""Generate a unique machine fingerprint"""
try:
hostname = socket.gethostname()
username = getpass.getuser()
unique_string = f"{hostname}-{username}-kiro-gateway"
return hashlib.sha256(unique_string.encode()).hexdigest()
except:
return hashlib.sha256(b"default-machine-fingerprint").hexdigest()
def generate_completion_id() -> str:
"""Generate a unique completion ID"""
return f"chatcmpl-{uuid_lib.uuid4().hex}"
def generate_conversation_id(messages: Optional[List[dict]] = None) -> str:
"""Generate a stable conversation ID from messages"""
if not messages:
return str(uuid_lib.uuid4())
# Use first 3 messages and last message for hashing
key_messages = messages[:3]
if len(messages) > 3:
key_messages.append(messages[-1])
# Create a stable string for hashing
content = ""
for msg in key_messages:
role = msg.get('role', '')
content_text = str(msg.get('content', ''))[:100]
content += f"{role}:{content_text[:50]}"
return hashlib.sha256(content.encode()).hexdigest()[:16]
def generate_tool_call_id() -> str:
"""Generate a unique ID for tool calls"""
return f"call_{uuid_lib.uuid4().hex[:8]}"
def get_kiro_headers(auth_manager, token: str) -> dict:
"""Get headers for Kiro API requests"""
fingerprint = get_machine_fingerprint()
return {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"User-Agent": f"aws-sdk-js/1.0.27 KiroIDE-0.7.45-{fingerprint}",
"x-amz-user-agent": f"aws-sdk-js/1.0.27 KiroIDE-0.7.45-{fingerprint}",
"x-amz-codewhisperer-optout": "true",
"x-amzn-kiro-agent-mode": "vibe",
"amz-sdk-invocation-id": str(uuid_lib.uuid4()),
"amz-sdk-request": "attempt=1; max=3"
}
def normalize_model_name(model_name: str) -> str:
"""Normalize model name for Kiro API"""
# Convert various model name formats to Kiro's expected format
model_map = {
'claude-sonnet-4-5': 'claude-sonnet-4-5',
'claude-sonnet-4.5': 'claude-sonnet-4-5',
'claude-sonnet-4': 'claude-sonnet-4',
'claude-haiku-4-5': 'claude-haiku-4-5',
'claude-haiku-4.5': 'claude-haiku-4-5',
'claude-opus-4-5': 'claude-opus-4-5',
'claude-opus-4.5': 'claude-opus-4-5',
'claude-3-5-sonnet': 'claude-sonnet-4-5',
'claude-3-5-sonnet': 'claude-sonnet-4-5',
}
# Normalize the model name
model_lower = model_name.lower().replace('_', '-')
# Check for known aliases
for alias, normalized in model_map.items():
if model_lower == alias or model_lower == alias.replace('-', '_'):
return normalized
# If not in map, try to normalize common patterns
if 'sonnet' in model_lower and '4.5' in model_lower:
return 'claude-sonnet-4-5'
elif 'haiku' in model_lower and '4.5' in model_lower:
return 'claude-haiku-4-5'
elif 'opus' in model_lower and '4.5' in model_lower:
return 'claude-opus-4-5'
elif 'sonnet' in model_lower and '4' in model_lower:
return 'claude-sonnet-4'
# Default to the original name if no match
return model_lower
def build_kiro_request(messages: list, model: str, max_tokens: int = None,
temperature: float = 1.0, stream: bool = False) -> dict:
"""Build a Kiro API request from OpenAI-style request"""
# Convert messages to Kiro format
kiro_messages = []
for msg in messages:
role = msg.get('role')
content = msg.get('content', '')
if role == 'system':
# System messages need special handling
kiro_messages.append({
"role": "user",
"content": content
})
elif role in ['user', 'assistant', 'system']:
kiro_messages.append({
"role": role,
"content": content
})
elif role == 'tool':
# Tool messages need special handling
kiro_messages.append({
"role": "user",
"content": f"[Tool result: {content}]"
})
request = {
"model": normalize_model_name(model),
"messages": kiro_messages,
"max_tokens": max_tokens,
"temperature": temperature,
"stream": stream
}
return request
def parse_kiro_response(response_data: dict) -> dict:
"""Parse Kiro API response to OpenAI format"""
if not response_data:
return {
"id": f"kiro-{uuid_lib.uuid4().hex}",
"object": "chat.completion",
"created": 0,
"model": "unknown",
"choices": [],
"usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
}
# Extract the response
choices = []
if 'choices' in response_data:
for choice in response_data.get('choices', []):
message = choice.get('message', {})
choices.append({
"index": choice.get('index', 0),
"message": {
"role": "assistant",
"content": message.get('content', ''),
"tool_calls": message.get('tool_calls', [])
},
"finish_reason": choice.get('finish_reason', 'stop')
})
return {
"id": response_data.get('id', f"kiro-{uuid_lib.uuid4().hex}"),
"object": "chat.completion",
"created": response_data.get('created', 0),
"model": response_data.get('model', 'unknown'),
"choices": choices,
"usage": response_data.get('usage', {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0
})
}
\ No newline at end of file
aisbf/providers.py
View file @ b6bbf540
......@@ -823,7 +823,7 @@ class GoogleProviderHandler(BaseProviderHandler):
"id": f"google-{model}-{int(time.time())}",
"object": "chat.completion",
"created": int(time.time()),
"model": model,
"model": f"{self.provider_id}/{model}",
"choices": [{
"index": 0,
"message": {
......@@ -1141,7 +1141,7 @@ class AnthropicProviderHandler(BaseProviderHandler):
"id": f"anthropic-{model}-{int(time.time())}",
"object": "chat.completion",
"created": int(time.time()),
"model": model,
"model": f"{self.provider_id}/{model}",
"choices": [{
"index": 0,
"message": {
......@@ -1199,6 +1199,255 @@ class AnthropicProviderHandler(BaseProviderHandler):
Model(id="claude-3-opus-20240229", name="Claude 3 Opus", provider_id=self.provider_id)
]
class KiroProviderHandler(BaseProviderHandler):
"""
Handler for direct Kiro API integration (Amazon Q Developer).
This handler makes direct API calls to Kiro's API using credentials from
Kiro IDE or kiro-cli, with FULL kiro-gateway feature parity including:
- Tool calls/function calling
- Images/multimodal content
- Complex message merging and validation
- Role normalization
- Complete OpenAI <-> Kiro format conversion
"""
def __init__(self, provider_id: str, api_key: str):
super().__init__(provider_id, api_key)
self.provider_config = config.get_provider(provider_id)
self.region = "us-east-1" # Default region
# Initialize KiroAuthManager with credentials from config
self.auth_manager = None
self._init_auth_manager()
# HTTP client for making requests
self.client = httpx.AsyncClient(timeout=httpx.Timeout(300.0, connect=30.0))
def _init_auth_manager(self):
"""Initialize KiroAuthManager with credentials from config"""
try:
from .kiro_auth import KiroAuthManager
# Get Kiro-specific configuration from provider config
kiro_config = getattr(self.provider_config, 'kiro_config', None)
if not kiro_config:
import logging
logging.warning(f"No kiro_config found in provider {self.provider_id}, using defaults")
kiro_config = {}
# Extract credentials from provider config
refresh_token = kiro_config.get('refresh_token') if isinstance(kiro_config, dict) else None
profile_arn = kiro_config.get('profile_arn') if isinstance(kiro_config, dict) else None
region = kiro_config.get('region', 'us-east-1') if isinstance(kiro_config, dict) else 'us-east-1'
creds_file = kiro_config.get('creds_file') if isinstance(kiro_config, dict) else None
sqlite_db = kiro_config.get('sqlite_db') if isinstance(kiro_config, dict) else None
client_id = kiro_config.get('client_id') if isinstance(kiro_config, dict) else None
client_secret = kiro_config.get('client_secret') if isinstance(kiro_config, dict) else None
self.region = region
# Initialize auth manager
self.auth_manager = KiroAuthManager(
refresh_token=refresh_token,
profile_arn=profile_arn,
region=region,
creds_file=creds_file,
sqlite_db=sqlite_db,
client_id=client_id,
client_secret=client_secret
)
import logging
logging.info(f"KiroProviderHandler: Auth manager initialized for region {region}")
except Exception as e:
import logging
logging.error(f"Failed to initialize KiroAuthManager: {e}")
self.auth_manager = None
async def handle_request(self, model: str, messages: List[Dict], max_tokens: Optional[int] = None,
temperature: Optional[float] = 1.0, stream: Optional[bool] = False,
tools: Optional[List[Dict]] = None, tool_choice: Optional[Union[str, Dict]] = None) -> Union[Dict, object]:
if self.is_rate_limited():
raise Exception("Provider rate limited")
try:
import logging
import json
import uuid
logging.info(f"KiroProviderHandler: Handling request for model {model}")
if AISBF_DEBUG:
logging.info(f"KiroProviderHandler: Messages: {messages}")
logging.info(f"KiroProviderHandler: Tools: {tools}")
else:
logging.info(f"KiroProviderHandler: Messages count: {len(messages)}")
logging.info(f"KiroProviderHandler: Tools count: {len(tools) if tools else 0}")
if not self.auth_manager:
raise Exception("Kiro authentication not configured. Please set kiro_config in provider configuration.")
# Apply rate limiting
await self.apply_rate_limit()
# Get access token and profile ARN
access_token = await self.auth_manager.get_access_token()
profile_arn = self.auth_manager.profile_arn
if not profile_arn:
raise Exception("Profile ARN not available. Please configure Kiro credentials.")
# Use full kiro-gateway conversion pipeline
from .kiro_converters_openai import build_kiro_payload_from_dict
conversation_id = str(uuid.uuid4())
# Build Kiro API payload using full conversion pipeline
# This handles ALL features: tools, images, message merging, role normalization, etc.
payload = build_kiro_payload_from_dict(
model=model,
messages=messages,
tools=tools,
conversation_id=conversation_id,
profile_arn=profile_arn
)
if AISBF_DEBUG:
logging.info(f"KiroProviderHandler: Kiro payload: {json.dumps(payload, indent=2)}")
# Make request to Kiro API
headers = self.auth_manager.get_auth_headers(access_token)
headers["Content-Type"] = "application/json"
kiro_api_url = f"https://q.{self.region}.amazonaws.com/generateAssistantResponse"
logging.info(f"KiroProviderHandler: Sending request to {kiro_api_url}")
response = await self.client.post(
kiro_api_url,
json=payload,
headers=headers
)
response.raise_for_status()
response_data = response.json()
if AISBF_DEBUG:
logging.info(f"KiroProviderHandler: Raw Kiro response: {json.dumps(response_data, indent=2)}")
logging.info(f"KiroProviderHandler: Response received")
# Parse Kiro response and convert to OpenAI format
openai_response = self._parse_kiro_response(response_data, model)
self.record_success()
return openai_response
except Exception as e:
import logging
logging.error(f"KiroProviderHandler: Error: {str(e)}", exc_info=True)
self.record_failure()
raise e
def _parse_kiro_response(self, kiro_response: Dict, model: str) -> Dict:
"""
Parse Kiro API response and convert to OpenAI format.
Handles:
- Text content
- Tool calls (toolUses)
- Finish reasons
- Usage statistics
"""
import logging
import json
# Extract assistant message content
assistant_content = ""
tool_calls = None
finish_reason = "stop"
# Kiro response structure varies, try different paths
if "message" in kiro_response:
assistant_content = kiro_response["message"]
elif "content" in kiro_response:
assistant_content = kiro_response["content"]
elif "conversationState" in kiro_response:
conv_state = kiro_response["conversationState"]
if "currentMessage" in conv_state:
current_msg = conv_state["currentMessage"]
if "assistantResponseMessage" in current_msg:
assistant_msg = current_msg["assistantResponseMessage"]
assistant_content = assistant_msg.get("content", "")
# Check for tool uses
if "toolUses" in assistant_msg:
tool_uses = assistant_msg["toolUses"]
tool_calls = []
for idx, tool_use in enumerate(tool_uses):
tool_call = {
"id": tool_use.get("toolUseId", f"call_{idx}"),
"type": "function",
"function": {
"name": tool_use.get("name", ""),
"arguments": json.dumps(tool_use.get("input", {}))
}
}
tool_calls.append(tool_call)
if tool_calls:
finish_reason = "tool_calls"
logging.info(f"KiroProviderHandler: Parsed {len(tool_calls)} tool calls from response")
# Build OpenAI-style response
openai_response = {
"id": f"kiro-{int(time.time())}",
"object": "chat.completion",
"created": int(time.time()),
"model": f"{self.provider_id}/{model}",
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": assistant_content if not tool_calls else None
},
"finish_reason": finish_reason
}],
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0
}
}
# Add tool_calls if present
if tool_calls:
openai_response["choices"][0]["message"]["tool_calls"] = tool_calls
return openai_response
async def get_models(self) -> List[Model]:
try:
import logging
logging.info("KiroProviderHandler: Getting models list")
# Apply rate limiting
await self.apply_rate_limit()
# Return static list of Claude models available through Kiro
return [
Model(id="anthropic.claude-3-5-sonnet-20241022-v2:0", name="Claude 3.5 Sonnet v2", provider_id=self.provider_id),
Model(id="anthropic.claude-3-5-haiku-20241022-v1:0", name="Claude 3.5 Haiku", provider_id=self.provider_id),
Model(id="anthropic.claude-3-5-sonnet-20240620-v1:0", name="Claude 3.5 Sonnet v1", provider_id=self.provider_id),
Model(id="anthropic.claude-sonnet-3-5-v2", name="Claude 3.5 Sonnet v2 (alias)", provider_id=self.provider_id),
Model(id="claude-sonnet-4-5", name="Claude 3.5 Sonnet v2 (short)", provider_id=self.provider_id),
Model(id="claude-haiku-4-5", name="Claude 3.5 Haiku (short)", provider_id=self.provider_id),
]
except Exception as e:
import logging
logging.error(f"KiroProviderHandler: Error getting models: {str(e)}", exc_info=True)
raise e
class OllamaProviderHandler(BaseProviderHandler):
def __init__(self, provider_id: str, api_key: Optional[str] = None):
super().__init__(provider_id, api_key)
......@@ -1335,7 +1584,7 @@ class OllamaProviderHandler(BaseProviderHandler):
"id": f"ollama-{model}-{int(time.time())}",
"object": "chat.completion",
"created": int(time.time()),
"model": model,
"model": f"{self.provider_id}/{model}",
"choices": [{
"index": 0,
"message": {
......@@ -1375,7 +1624,8 @@ PROVIDER_HANDLERS = {
'google': GoogleProviderHandler,
'openai': OpenAIProviderHandler,
'anthropic': AnthropicProviderHandler,
'ollama': OllamaProviderHandler
'ollama': OllamaProviderHandler,
'kiro': KiroProviderHandler
}
def get_provider_handler(provider_id: str, api_key: Optional[str] = None) -> BaseProviderHandler:
......
config/providers.json
View file @ b6bbf540
......@@ -180,6 +180,32 @@
"type": "microsoft",
"api_key_required": true,
"rate_limit": 0
},
"kiro": {
"id": "kiro",
"name": "Kiro IDE (Amazon Q Developer)",
"endpoint": "https://q.us-east-1.amazonaws.com",
"type": "kiro",
"api_key_required": false,
"rate_limit": 0,
"kiro_config": {
"_comment": "Uses Kiro IDE credentials (VS Code extension)",
"creds_file": "~/.config/Code/User/globalStorage/amazon.q/credentials.json",
"region": "us-east-1"
}
},
"kiro-cli": {
"id": "kiro-cli",
"name": "Kiro CLI (Amazon Q Developer)",
"endpoint": "https://q.us-east-1.amazonaws.com",
"type": "kiro",
"api_key_required": false,
"rate_limit": 0,
"kiro_config": {
"_comment": "Uses kiro-cli credentials (SQLite database)",
"sqlite_db": "~/.local/share/kiro-cli/data.sqlite3",
"region": "us-east-1"
}
}
}
}
config/rotations.json
View file @ b6bbf540
......@@ -144,6 +144,45 @@
]
}
]
},
"kiro-claude": {
"model_name": "kiro-claude",
"notifyerrors": false,
"providers": [
{
"provider_id": "kiro",
"api_key": "YOUR_KIRO_GATEWAY_API_KEY",
"models": [
{
"name": "claude-sonnet-4-5",
"weight": 3,
"rate_limit": 0,
"max_request_tokens": 200000,
"context_size": 200000,
"condense_context": 80,
"condense_method": ["hierarchical", "semantic"]
},
{
"name": "claude-haiku-4-5",
"weight": 2,
"rate_limit": 0,
"max_request_tokens": 200000,
"context_size": 200000,
"condense_context": 75,
"condense_method": "conversational"
},
{
"name": "claude-sonnet-4",
"weight": 1,
"rate_limit": 0,
"max_request_tokens": 200000,
"context_size": 200000,
"condense_context": 80,
"condense_method": "conversational"
}
]
}
]
}
}
}
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment