feat: a2a & extension

Author: meguminnnnnnnnn

a2a/client/README.md

@@ -0,0 +1,155 @@
+# A2A Client (Go)
+
+This package provides a Go client implementation for the Agent-to-Agent (A2A) communication protocol.
+
+## Features
+
+- JSON-RPC 2.0 compliant client
+- Supports core A2A methods:
+  - `tasks/send`: Send a new task
+  - `tasks/get`: Get task status
+  - `tasks/cancel`: Cancel a task
+- Streaming task updates with Server-Sent Events (SSE)
+- Error handling with A2A error codes
+- Type-safe request/response handling
+
+## Usage
+
+```go
+package main
+
+import (
+    "log"
+    "a2a/client"
+    "a2a/models"
+)
+
+func main() {
+    // Create a new client
+    a2aClient := client.NewClient("http://localhost:8080")
+
+    // Create a task message
+    message := models.Message{
+        Role: "user",
+        Parts: []models.Part{
+            {
+                Type: stringPtr("text"),
+                Text: stringPtr("Hello, A2A agent!"),
+            },
+        },
+    }
+
+    // Send a task
+    response, err := a2aClient.SendTask(models.TaskSendParams{
+        ID:      "task-1",
+        Message: message,
+    })
+    if err != nil {
+        log.Fatalf("Failed to send task: %v", err)
+    }
+
+    // Get the task from the response
+    task, ok := response.Result.(*models.Task)
+    if !ok {
+        log.Fatalf("Expected result to be a Task")
+    }
+
+    // Use the task...
+}
+```
+
+## API
+
+### NewClient
+
+```go
+func NewClient(baseURL string) *Client
+```
+
+Creates a new A2A client instance with the specified base URL.
+
+### Client Methods
+
+#### SendTask
+
+```go
+func (c *Client) SendTask(params models.TaskSendParams) (*models.JSONRPCResponse, error)
+```
+
+Sends a new task to the agent. Returns a JSON-RPC response containing the task or an error.
+
+#### GetTask
+
+```go
+func (c *Client) GetTask(params models.TaskQueryParams) (*models.JSONRPCResponse, error)
+```
+
+Gets the status of a task. Returns a JSON-RPC response containing the task or an error.
+
+#### CancelTask
+
+```go
+func (c *Client) CancelTask(params models.TaskIDParams) (*models.JSONRPCResponse, error)
+```
+
+Cancels a task. Returns a JSON-RPC response containing the task or an error.
+
+## Streaming Support
+
+The client supports streaming task updates using Server-Sent Events (SSE). To use streaming:
+
+1. Set the `Accept` header to `text/event-stream` in your request
+2. The server will respond with a stream of task status updates
+3. Each update will be a JSON object containing:
+   - Task ID
+   - Current status
+   - Whether it's the final update
+
+Example streaming usage:
+```go
+// Create a task with streaming
+message := models.Message{
+    Role: "user",
+    Parts: []models.Part{
+        {
+            Type: stringPtr("text"),
+            Text: stringPtr("Hello, A2A agent!"),
+        },
+    },
+}
+
+// Send a task with streaming enabled
+response, err := a2aClient.SendTaskWithStreaming(models.TaskSendParams{
+    ID:      "task-1",
+    Message: message,
+})
+if err != nil {
+    log.Fatalf("Failed to send task: %v", err)
+}
+
+// Process streaming updates
+for update := range response.Updates {
+    if update.Error != nil {
+        log.Printf("Error: %v", update.Error)
+        continue
+    }
+    
+    // Process the update
+    log.Printf("Task %s: %s", update.Result.ID, update.Result.Status.State)
+}
+```
+
+## Testing
+
+Run the tests with:
+
+```bash
+go test ./...
+```
+
+The test suite includes examples of:
+- Sending tasks
+- Getting task status
+- Canceling tasks
+- Streaming task updates
+- Error handling 

a2a/client/client.go

@@ -0,0 +1,92 @@
+package client
+
+import (
+	"context"
+	"errors"
+
+	"github.com/cloudwego/eino-ext/a2a/models"
+	"github.com/cloudwego/eino-ext/a2a/transport"
+)
+
+// A2AClient represents an A2A protocol client
+type A2AClient struct {
+	cli transport.ClientTransport
+}
+
+type Config struct {
+	Transport transport.ClientTransport
+}
+
+// NewA2AClient creates a new A2A client
+func NewA2AClient(ctx context.Context, config *Config) (*A2AClient, error) {
+	if config == nil {
+		return nil, errors.New("config is required")
+	}
+
+	return &A2AClient{cli: config.Transport}, nil
+}
+
+func (c *A2AClient) AgentCard(ctx context.Context) (*models.AgentCard, error) {
+	return c.cli.AgentCard(ctx)
+}
+
+func (c *A2AClient) SendMessage(ctx context.Context, params *models.MessageSendParams) (*models.SendMessageResponseUnion, error) {
+	if params != nil {
+		(&params.Message).EnsureRequiredFields()
+	}
+	return c.cli.SendMessage(ctx, params)
+}
+
+func (c *A2AClient) SendMessageStreaming(ctx context.Context, params *models.MessageSendParams) (*ServerStreamingWrapper, error) {
+	if params != nil {
+		(&params.Message).EnsureRequiredFields()
+	}
+	stream, err := c.cli.SendMessageStreaming(ctx, params)
+	if err != nil {
+		return nil, err
+	}
+	return &ServerStreamingWrapper{stream}, nil
+}
+
+func (c *A2AClient) GetTask(ctx context.Context, params *models.TaskQueryParams) (*models.Task, error) {
+	return c.cli.GetTask(ctx, params)
+}
+
+func (c *A2AClient) CancelTask(ctx context.Context, params *models.TaskIDParams) (*models.Task, error) {
+	return c.cli.CancelTask(ctx, params)
+}
+
+func (c *A2AClient) ResubscribeTask(ctx context.Context, params *models.TaskIDParams) (*ServerStreamingWrapper, error) {
+	stream, err := c.cli.ResubscribeTask(ctx, params)
+	if err != nil {
+		return nil, err
+	}
+	return &ServerStreamingWrapper{stream}, nil
+}
+
+func (c *A2AClient) SetPushNotificationConfig(ctx context.Context, params *models.TaskPushNotificationConfig) (*models.TaskPushNotificationConfig, error) {
+	return c.cli.SetPushNotificationConfig(ctx, params)
+}
+
+func (c *A2AClient) GetPushNotificationConfig(ctx context.Context, params *models.GetTaskPushNotificationConfigParams) (*models.TaskPushNotificationConfig, error) {
+	return c.cli.GetPushNotificationConfig(ctx, params)
+}
+
+// todo: list/delete notification config & agent/authenticatedExtendedCard
+
+type ServerStreamingWrapper struct {
+	s models.ResponseReader
+}
+
+func (s *ServerStreamingWrapper) Recv() (resp *models.SendMessageStreamingResponseUnion, err error) {
+	defer func() {
+		if err != nil {
+			s.s.Close()
+		}
+	}()
+	return s.s.Read()
+}
+
+func (s *ServerStreamingWrapper) Close() error {
+	return s.s.Close()
+}

a2a/client/client_test.go

@@ -0,0 +1 @@
+package client