Skip to main content

PAF-Core Overview

Pixell Agent Framework Core (PAF-Core) is the foundational protocol and framework layer that enables agent-to-agent (A2A) communication in the Pixell ecosystem. PAF-Core defines the standardized communication protocols, message formats, and service discovery mechanisms that allow agents to discover, communicate, and collaborate with each other.

What is PAF-Core?​

PAF-Core is not a standalone application—it's the shared protocol specification and framework components implemented across PAK and PAR that enable:

  • ✅ A2A Protocol - gRPC-based Agent-to-Agent communication
  • ✅ Service Discovery - AWS Cloud Map integration for finding agents
  • ✅ Message Routing - Intelligent routing between agents
  • ✅ Protocol Buffers - Standardized message formats
  • ✅ Security Model - Authentication and authorization framework
  • ✅ Intent System - Capability discovery and invocation (planned)

Architecture​

A2A (Agent-to-Agent) Protocol​

The A2A protocol is the core of PAF-Core, enabling direct agent-to-agent communication.

Protocol Definition​

PAF-Core uses gRPC with Protocol Buffers for efficient, type-safe communication:

syntax = "proto3";

package pixell.agent;

// Agent service definition
service AgentService {
  // Health check
  rpc Health(Empty) returns (HealthStatus);
  
  // Describe agent capabilities
  rpc DescribeCapabilities(Empty) returns (Capabilities);
  
  // Invoke an action
  rpc Invoke(ActionRequest) returns (ActionResult);
  
  // Simple ping
  rpc Ping(Empty) returns (Pong);
}

// Action request message
message ActionRequest {
  string action = 1;
  map<string, string> parameters = 2;
  string request_id = 3;
}

// Action result message
message ActionResult {
  bool success = 1;
  string result = 2;
  string error = 3;
  string request_id = 4;
  int64 duration_ms = 5;
}

Service Methods​

Health()​

Health check to verify agent availability:

Request:

{}

Response:

{
  "ok": true,
  "message": "healthy",
  "timestamp": 1704067200000
}

DescribeCapabilities()​

Discover what actions an agent can perform:

Request:

{}

Response:

{
  "methods": ["process", "analyze", "transform"],
  "metadata": {
    "version": "1.0.0",
    "description": "Data processing agent"
  }
}

Invoke()​

Execute an agent action:

Request:

{
  "action": "process",
  "parameters": {
    "input": "data",
    "format": "json"
  },
  "request_id": "req_123"
}

Response:

{
  "success": true,
  "result": "{\"processed\": true}",
  "request_id": "req_123",
  "duration_ms": 150
}

Ping()​

Simple connectivity test:

Request:

{}

Response:

{
  "message": "pong",
  "timestamp": 1704067200000
}

Service Discovery​

PAF-Core implements hybrid service discovery for different deployment scenarios:

Internal Discovery (AWS Cloud Map)​

For agents within the same VPC:

from pixell_runtime.a2a.client import get_a2a_client

# Client automatically uses service discovery
client = get_a2a_client(prefer_internal=True)

# Discover agent by ID
channel = client.get_agent_channel(deployment_id="agent-123")

# Or discover any healthy agent
channel = client.get_agent_channel()

External Discovery (Network Load Balancer)​

For external clients:

import os

# Set external endpoint
os.environ['A2A_EXTERNAL_ENDPOINT'] = 'runtime-nlb.amazonaws.com:50051'

# Client uses NLB endpoint
client = get_a2a_client(prefer_internal=False)
channel = client.get_agent_channel()

Discovery Flow​

Message Routing​

PAF-Core provides intelligent message routing:

Direct Routing​

For known agent endpoints:

# Direct connection via IP and port
channel = grpc.insecure_channel("10.0.1.100:50051")

Service Discovery Routing​

For dynamic agent discovery:

# Automatic routing via service discovery
client = get_a2a_client()
channel = client.get_agent_channel(deployment_id="agent-123")

Load Balancing​

For multiple agent instances:

# Get list of healthy agents
agents = sd_client.discover_agents(max_results=5)

# Simple round-robin (implement your own strategy)
agent = agents[request_count % len(agents)]
channel = grpc.insecure_channel(f"{agent['ipv4']}:{agent['port']}")

Security Model​

PAF-Core provides a security framework (implementation in progress):

Authentication​

  • OIDC Tokens: Agent identity via OpenID Connect (planned)
  • mTLS: Mutual TLS for agent-to-agent communication (planned)
  • API Keys: Simple key-based authentication (current)

Authorization​

  • Capability-based: Agents declare capabilities, clients request specific actions
  • Rate Limiting: Prevent abuse via rate limits
  • Policy Engine: Define what agents can communicate with each other (planned)

Encryption​

  • TLS in Transit: All A2A communication over TLS
  • At Rest: Agent packages encrypted in storage

Intent System (Planned)​

The intent system will provide higher-level capability discovery:

# Future intent-based routing
from pixell.intent import IntentRouter

router = IntentRouter()

# Find agent that can handle intent
agent = router.find_agent_for_intent("process.data.json")

# Invoke via intent
result = await router.invoke_intent(
    intent="process.data.json",
    parameters={"input": data}
)

Intent Format​

# agent.yaml
intents:
  - name: process.data.json
    description: "Process JSON data"
    parameters:
      input:
        type: string
        required: true
      format:
        type: string
        default: "compact"
    returns:
      type: object

LangGraph Integration​

PAF-Core is designed to work with LangGraph for complex agent workflows:

from langgraph.graph import StateGraph
from pixell_runtime.a2a.client import get_a2a_client

# Create LangGraph workflow
workflow = StateGraph()

# Add A2A agent as node
async def call_pixell_agent(state):
    client = get_a2a_client()
    channel = client.get_agent_channel(deployment_id="processor")
    
    stub = agent_pb2_grpc.AgentServiceStub(channel)
    response = await stub.Invoke(
        agent_pb2.ActionRequest(
            action="process",
            parameters=state["data"]
        )
    )
    
    return {"result": response.result}

workflow.add_node("pixell_agent", call_pixell_agent)
workflow.set_entry_point("pixell_agent")

# Compile and run
app = workflow.compile()
result = await app.ainvoke({"data": {"input": "test"}})

Protocol Implementations​

PAF-Core is implemented across the Pixell stack:

PAK (Agent Kit)​

  • Protocol validation during build
  • Intent definition in agent.yaml
  • Client code generation

PAR (Agent Runtime)​

  • A2A gRPC server implementation
  • Service discovery client
  • Message routing
  • Protocol buffer compilation

Agent Code​

  • Service implementation
  • Client libraries
  • Error handling

Example: Agent-to-Agent Communication​

Complete example of A2A communication:

Agent A (Client)​

from pixell_runtime.a2a.client import get_a2a_client
from pixell_runtime.proto import agent_pb2, agent_pb2_grpc

async def call_agent_b():
    # Get client
    client = get_a2a_client()
    
    # Get channel to Agent B
    channel = client.get_agent_channel(deployment_id="agent-b")
    
    # Create stub
    stub = agent_pb2_grpc.AgentServiceStub(channel)
    
    # Call Agent B
    response = await stub.Invoke(
        agent_pb2.ActionRequest(
            action="analyze",
            parameters={"data": "sample input"}
        )
    )
    
    if response.success:
        print(f"Result: {response.result}")
    else:
        print(f"Error: {response.error}")

Agent B (Server)​

from pixell_runtime.proto import agent_pb2, agent_pb2_grpc

class MyAgentService(agent_pb2_grpc.AgentServiceServicer):
    async def Invoke(self, request, context):
        action = request.action
        params = dict(request.parameters)
        
        if action == "analyze":
            # Perform analysis
            result = analyze_data(params.get("data"))
            
            return agent_pb2.ActionResult(
                success=True,
                result=result,
                request_id=request.request_id
            )
        else:
            return agent_pb2.ActionResult(
                success=False,
                error=f"Unknown action: {action}",
                request_id=request.request_id
            )

Performance Characteristics​

PAF-Core is designed for high-performance communication:

  • Latency: < 100ms for A2A calls within same VPC
  • Throughput: 10,000+ messages/second per agent
  • Concurrency: Handles thousands of concurrent connections
  • Reliability: Automatic retry and failover

Next Steps​

Learn More​

  • Protocol Buffers: See pixell_runtime/proto/agent.proto
  • Client Implementation: See pixell_runtime/a2a/client.py
  • Server Implementation: See pixell_runtime/a2a/server.py
  • Service Discovery: See pixell_runtime/utils/service_discovery.py