Snowflake-Labs/cortex_agent_a2a

Snowflake-Labs/cortex_agent_a2a

Language: Python

License: Apache-2.0

Stars: 2

Forks: 0

Open issues: 0

Created: 2026-04-24T20:11:13Z

Pushed: 2026-04-25T01:41:03Z

Default branch: main

Fork: no

Archived: no

README:

🔷 Snowflake Cortex A2A Agent

An Agent-to-Agent (A2A) protocol wrapper for any Snowflake Cortex Agent. This service exposes your Cortex Agent through the Google A2A protocol, enabling other AI agents to interact with it through a standardized interface.

🏗️ Architecture

┌─────────────────┐ ┌──────────────────────┐ ┌─────────────────────┐
│ A2A Client │────▶│ A2A Wrapper │────▶│ Snowflake Cortex │
│ (Other Agents) │ │ (This Service) │ │ Agent │
└─────────────────┘ └──────────────────────┘ └─────────────────────┘
│
├── auth.py (JWT Authentication)
├── executor.py (Cortex Integration)
└── main.py (A2A Server)

Components:

• A2A SDK: Handles HTTP server, JSON-RPC routing, and protocol compliance
• Agent Executor: Custom logic that receives A2A tasks, calls Snowflake, and returns responses
• Snowflake Cortex: Your backend Cortex Agent (configurable via environment variables)

📋 Prerequisites

• Python 3.11+
• Snowflake account with Cortex Agent access
• RSA key pair for Snowflake authentication
• A deployed Cortex Agent in Snowflake

🚀 Quick Start

1. Clone and Setup Environment

cd cortex_agent_a2a
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt

2. Generate RSA Key Pair (if needed)

# Generate private key
openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt

# Generate public key
openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub

# Get the public key content (for Snowflake)
grep -v "BEGIN\|END" rsa_key.pub | tr -d '\n'

3. Configure Snowflake User

Run this SQL in Snowflake (replace with your public key):

ALTER USER your_username SET RSA_PUBLIC_KEY='MIIBIjANBgkq...your_public_key...AQAB';

-- Verify the key is set
DESC USER your_username;
-- Look for RSA_PUBLIC_KEY_FP

4. Configure Environment

Create a .env file based on env.template:

# Snowflake Connection
SNOWFLAKE_ACCOUNT_LOCATOR=YOUR_ACCOUNT_LOCATOR
SNOWFLAKE_ACCOUNT=YOUR_ORG-YOUR_ACCOUNT
SNOWFLAKE_USER=your_username
PRIVATE_KEY_PATH=rsa_key.p8

# Cortex Agent Details - Point to YOUR agent
AGENT_DATABASE=YOUR_DATABASE
AGENT_SCHEMA=YOUR_SCHEMA
AGENT_NAME=YOUR_CORTEX_AGENT

# Optional: Customize descriptions
AGENT_DESCRIPTION=My custom Cortex Agent description
AGENT_URL=http://localhost:8000

Finding your account identifiers:

-- Get account locator (for JWT)
SELECT CURRENT_ACCOUNT();

-- Get full account identifier (for URL - replace underscores with hyphens)
SELECT CURRENT_ORGANIZATION_NAME() || '-' || CURRENT_ACCOUNT_NAME();

5. Run the Server

source venv/bin/activate
python main.py

The server will start on http://localhost:8000

🔌 API Endpoints

Discovery Endpoint

GET /.well-known/agent.json

Returns the Agent Card describing capabilities:

{
"name": "Cortex Agent: YOUR_AGENT_NAME",
"description": "Your agent description...",
"version": "1.0.0",
"skills": [
{
"id": "query_cortex_agent",
"name": "Cortex Agent Query",
"description": "Sends queries to the Snowflake Cortex Agent...",
"tags": ["snowflake", "cortex", "ai", "analytics"]
}
],
"capabilities": {
"streaming": false,
"pushNotifications": false
}
}

Task Endpoint (JSON-RPC)

POST /
Content-Type: application/json

Request:

{
"jsonrpc": "2.0",
"method": "message/send",
"id": "unique-request-id",
"params": {
"message": {
"messageId": "unique-message-id",
"role": "user",
"parts": [
{
"type": "text",
"text": "Your question here"
}
]
}
}
}

Response:

{
"jsonrpc": "2.0",
"id": "unique-request-id",
"result": {
"kind": "message",
"messageId": "response-message-id",
"role": "agent",
"parts": [
{
"kind": "text",
"text": "Agent's response..."
}
]
}
}

📝 Example Usage

Using cURL

# Check agent discovery
curl http://localhost:8000/.well-known/agent.json

# Send a query to the Cortex Agent
curl -X POST http://localhost:8000/ \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"method": "message/send",
"id": "1",
"params": {
"message": {
"messageId": "msg-001",
"role": "user",
"parts": [{"type": "text", "text": "What data do you have access to?"}]
}
}
}'

Using Python

import requests

# Discovery
response = requests.get("http://localhost:8000/.well-known/agent.json")
print(response.json())

# Query
response = requests.post(
"http://localhost:8000/",
json={
"jsonrpc": "2.0",
"method": "message/send",
"id": "1",
"params": {
"message": {
"messageId": "msg-001",
"role": "user",
"parts": [{"type": "text", "text": "Your question here"}]
}
}
}
)
print(response.json())

🧪 Testing with test_a2a.py

A lightweight test client is included to verify the A2A server is working correctly.

Basic Usage

# Make sure the server is running first
python main.py

# In another terminal, run the test client
python test_a2a.py

Command Line Options

| Option | Description | Default | |--------|-------------|---------| | --query "..." | The question to send to the agent | "What data do you have access to?" | | --url URL | Base URL of the A2A server | http://localhost:8000 | | --card-only | Only fetch the agent card, don't send a query | - |

Examples

# Send a custom query
python test_a2a.py --query "Show me all players from Real Madrid"

# Just check the agent card (discovery endpoint)
python test_a2a.py --card-only

# Test against a different server/port
python test_a2a.py --url http://localhost:8001 --query "Hello"

Sample Output

🔷 Snowflake Cortex A2A Agent Test Client
Server: http://localhost:8000

============================================================
📋 Fetching Agent Card...
============================================================
Name: Cortex Agent: YOUR_AGENT_NAME
Description: Your agent description
Version: 1.0.0
Skills: ['Cortex Agent Query']
Streaming: False

============================================================
📨 Sending Query: Show me players from Barcelona
============================================================

📥 Raw JSON Response:
{
"jsonrpc": "2.0",
"result": {
"kind": "message",
"parts": [{"kind": "text", "text": "..."}]
}
}…