Fix issue on context id generation (#45)

* fix issue on context id generation

* fix comment

* optimize code

Author: hyprh

examples/go.mod

@@ -9,7 +9,8 @@ require (
 	github.com/lestrrat-go/jwx/v2 v2.1.4
 	github.com/redis/go-redis/v9 v9.10.0
 	golang.org/x/oauth2 v0.29.0
-	trpc.group/trpc-go/trpc-a2a-go v0.0.0
+	trpc.group/trpc-go/trpc-a2a-go v0.0.3
+	trpc.group/trpc-go/trpc-a2a-go/taskmanager/redis v0.0.0-20250625115112-3bb198d0dc98
 )
 
 require (
@@ -31,3 +32,5 @@ require (
 )
 
 replace trpc.group/trpc-go/trpc-a2a-go => ../
+
+replace trpc.group/trpc-go/trpc-a2a-go/taskmanager/redis => ../taskmanager/redis

examples/simple/client/main.go

@@ -10,12 +10,11 @@ package main
 import (
 	"context"
 	"flag"
-	"log"
+	"fmt"
 	"time"
 
-	"github.com/google/uuid"
-
 	"trpc.group/trpc-go/trpc-a2a-go/client"
+	"trpc.group/trpc-go/trpc-a2a-go/log"
 	"trpc.group/trpc-go/trpc-a2a-go/protocol"
 )
 
@@ -24,6 +23,7 @@ func main() {
 	agentURL := flag.String("agent", "http://localhost:8080/", "Target A2A agent URL")
 	timeout := flag.Duration("timeout", 30*time.Second, "Request timeout (e.g., 30s, 1m)")
 	message := flag.String("message", "Hello, world!", "Message to send to the agent")
+	streaming := flag.Bool("streaming", false, "Use streaming mode (message/stream)")
 	flag.Parse()
 
 	// Create A2A client.
@@ -33,56 +33,105 @@ func main() {
 	}
 
 	// Display connection information.
-	log.Printf("Connecting to agent: %s (Timeout: %v)", *agentURL, *timeout)
-
-	// Create a new unique message ID and context ID.
-	contextID := uuid.New().String()
-	log.Printf("Context ID: %s", contextID)
+	log.Infof("Connecting to agent: %s (Timeout: %v)", *agentURL, *timeout)
+	log.Infof("Mode: %s", map[bool]string{true: "Streaming", false: "Standard"}[*streaming])
 
 	// Create the message to send using the new constructor.
-	userMessage := protocol.NewMessageWithContext(
+	userMessage := protocol.NewMessage(
 		protocol.MessageRoleUser,
 		[]protocol.Part{protocol.NewTextPart(*message)},
-		nil, // taskID
-		&contextID,
 	)
 
 	// Create message parameters using the new SendMessageParams structure.
 	params := protocol.SendMessageParams{
 		Message: userMessage,
 		Configuration: &protocol.SendMessageConfiguration{
-			Blocking: boolPtr(true), // Wait for completion
+			Blocking: boolPtr(false), // Non-blocking for streaming, blocking for standard
 		},
 	}
 
-	log.Printf("Sending message with content: %s", *message)
+	log.Infof("Sending message with content: %s", *message)
 
-	// Send message to the agent using the new message API.
+	// Create context for the request.
 	ctx, cancel := context.WithTimeout(context.Background(), *timeout)
 	defer cancel()
 
+	if *streaming {
+		// Use streaming mode
+		handleStreamingMode(ctx, a2aClient, params)
+	} else {
+		// Use standard mode
+		handleStandardMode(ctx, a2aClient, params)
+	}
+}
+
+// handleStreamingMode handles streaming message sending and event processing
+func handleStreamingMode(ctx context.Context, a2aClient *client.A2AClient, params protocol.SendMessageParams) {
+	log.Infof("Starting streaming request...")
+
+	// Send streaming message request
+	eventChan, err := a2aClient.StreamMessage(ctx, params)
+	if err != nil {
+		log.Fatalf("Failed to start streaming: %v", err)
+	}
+
+	log.Infof("Processing streaming events...")
+
+	eventCount := 0
+	var finalResult string
+
+	// Process streaming events
+	for {
+		select {
+		case event, ok := <-eventChan:
+			if !ok {
+				log.Infof("Stream completed. Total events received: %d", eventCount)
+				if finalResult != "" {
+					log.Infof("Final result: %s", finalResult)
+				}
+				return
+			}
+
+			eventCount++
+			log.Infof("Event %d received: %s", eventCount, getEventDescription(event))
+
+			// Extract final result from completed events
+			if result := extractFinalResult(event); result != "" {
+				finalResult = result
+				log.Infof("Received msg: [Text: %s]", result)
+			}
+
+		case <-ctx.Done():
+			log.Infof("Request timed out after receiving %d events", eventCount)
+			return
+		}
+	}
+}
+
+// handleStandardMode handles standard (non-streaming) message sending
+func handleStandardMode(ctx context.Context, a2aClient *client.A2AClient, params protocol.SendMessageParams) {
 	messageResult, err := a2aClient.SendMessage(ctx, params)
 	if err != nil {
 		log.Fatalf("Failed to send message: %v", err)
 	}
 
 	// Display the result.
-	log.Printf("Message sent successfully")
+	log.Infof("Message sent successfully")
 
 	// Handle the result based on its type
 	switch result := messageResult.Result.(type) {
 	case *protocol.Message:
-		log.Printf("Received message response:")
+		log.Infof("Received message response:")
 		printMessage(*result)
 	case *protocol.Task:
-		log.Printf("Received task response - ID: %s, State: %s", result.ID, result.Status.State)
+		log.Infof("Received task response - ID: %s, State: %s", result.ID, result.Status.State)
 
 		// If task is not completed, wait and check again
 		if result.Status.State != protocol.TaskStateCompleted &&
 			result.Status.State != protocol.TaskStateFailed &&
 			result.Status.State != protocol.TaskStateCanceled {
 
-			log.Printf("Task %s is %s, fetching final state...", result.ID, result.Status.State)
+			log.Infof("Task %s is %s, fetching final state...", result.ID, result.Status.State)
 
 			// Get the task's final state.
 			queryParams := protocol.TaskQueryParams{
@@ -97,65 +146,124 @@ func main() {
 				log.Fatalf("Failed to get task status: %v", err)
 			}
 
-			log.Printf("Task %s final state: %s", task.ID, task.Status.State)
+			log.Infof("Task %s final state: %s", task.ID, task.Status.State)
 			printTaskResult(task)
 		} else {
 			printTaskResult(result)
 		}
 	default:
-		log.Printf("Received unknown result type: %T", result)
+		log.Infof("Received unknown result type: %T", result)
+	}
+}
+
+// getEventDescription returns a human-readable description of the streaming event
+func getEventDescription(event protocol.StreamingMessageEvent) string {
+	switch result := event.Result.(type) {
+	case *protocol.Message:
+		ctxID := "unknown"
+		if result.ContextID != nil {
+			ctxID = *result.ContextID
+		}
+		return fmt.Sprintf("Message from %s, ContextID: %v", result.Role, ctxID)
+	case *protocol.Task:
+		return fmt.Sprintf("Task %s - State: %s, ContextID: %v", result.ID, result.Status.State, result.ContextID)
+	case *protocol.TaskStatusUpdateEvent:
+		return fmt.Sprintf(
+			"Status Update - Task: %s, State: %s, ContextID: %v",
+			result.TaskID,
+			result.Status.State,
+			result.ContextID,
+		)
+	case *protocol.TaskArtifactUpdateEvent:
+		artifactName := "Unnamed"
+		if result.Artifact.Name != nil {
+			artifactName = *result.Artifact.Name
+		}
+		return fmt.Sprintf("Artifact Update - %s, ContextID: %v", artifactName, result.ContextID)
+	default:
+		return fmt.Sprintf("Unknown event type: %T", result)
+	}
+}
+
+// extractFinalResult extracts the final text result from streaming events
+func extractFinalResult(event protocol.StreamingMessageEvent) string {
+	switch result := event.Result.(type) {
+	case *protocol.Message:
+		// Extract text from message parts
+		for _, part := range result.Parts {
+			if textPart, ok := part.(*protocol.TextPart); ok {
+				return textPart.Text
+			}
+		}
+	case *protocol.Task:
+		// Extract text from task status message
+		if result.Status.Message != nil {
+			for _, part := range result.Status.Message.Parts {
+				if textPart, ok := part.(*protocol.TextPart); ok {
+					return textPart.Text
+				}
+			}
+		}
+	case *protocol.TaskArtifactUpdateEvent:
+		// Extract text from artifact parts
+		for _, part := range result.Artifact.Parts {
+			if textPart, ok := part.(*protocol.TextPart); ok {
+				return textPart.Text
+			}
+		}
 	}
+	return ""
 }
 
 // printMessage prints the contents of a message.
 func printMessage(message protocol.Message) {
-	log.Printf("Message ID: %s", message.MessageID)
+	log.Infof("Message ID: %s", message.MessageID)
 	if message.ContextID != nil {
-		log.Printf("Context ID: %s", *message.ContextID)
+		log.Infof("Context ID: %s", *message.ContextID)
 	}
-	log.Printf("Role: %s", message.Role)
+	log.Infof("Role: %s", message.Role)
 
-	log.Printf("Message parts:")
+	log.Infof("Message parts:")
 	for i, part := range message.Parts {
 		switch p := part.(type) {
 		case *protocol.TextPart:
-			log.Printf("  Part %d (text): %s", i+1, p.Text)
+			log.Infof("  Part %d (text): %s", i+1, p.Text)
 		case *protocol.FilePart:
-			log.Printf("  Part %d (file): [file content]", i+1)
+			log.Infof("  Part %d (file): [file content]", i+1)
 		case *protocol.DataPart:
-			log.Printf("  Part %d (data): %+v", i+1, p.Data)
+			log.Infof("  Part %d (data): %+v", i+1, p.Data)
 		default:
-			log.Printf("  Part %d (unknown): %+v", i+1, part)
+			log.Infof("  Part %d (unknown): %+v", i+1, part)
 		}
 	}
 }
 
 // printTaskResult prints the contents of a task result.
 func printTaskResult(task *protocol.Task) {
 	if task.Status.Message != nil {
-		log.Printf("Task result message:")
+		log.Infof("Task result message:")
 		printMessage(*task.Status.Message)
 	}
 
 	// Print artifacts if any
 	if len(task.Artifacts) > 0 {
-		log.Printf("Task artifacts:")
+		log.Infof("Task artifacts:")
 		for i, artifact := range task.Artifacts {
 			name := "Unnamed"
 			if artifact.Name != nil {
 				name = *artifact.Name
 			}
-			log.Printf("  Artifact %d: %s", i+1, name)
+			log.Infof("  Artifact %d: %s", i+1, name)
 			for j, part := range artifact.Parts {
 				switch p := part.(type) {
 				case *protocol.TextPart:
-					log.Printf("    Part %d (text): %s", j+1, p.Text)
+					log.Infof("    Part %d (text): %s", j+1, p.Text)
 				case *protocol.FilePart:
-					log.Printf("    Part %d (file): [file content]", j+1)
+					log.Infof("    Part %d (file): [file content]", j+1)
 				case *protocol.DataPart:
-					log.Printf("    Part %d (data): %+v", j+1, p.Data)
+					log.Infof("    Part %d (data): %+v", j+1, p.Data)
 				default:
-					log.Printf("    Part %d (unknown): %+v", j+1, part)
+					log.Infof("    Part %d (unknown): %+v", j+1, part)
 				}
 			}
 		}