Skip to content

Commit 7ad5567

Browse files
EItanyaCopilotpeterj
authored
feat: Add BYO langgraph agents (#813)
This PR adds the ability to run langgraph agents as BYO agents in `kagent`. 1. Add `kagent` `checkpointer` implementation to the controller 2. Create `kagent-langgraph` python library to spin up agent --------- Signed-off-by: Eitan Yarmush <eitan.yarmush@solo.io> Signed-off-by: Peter Jausovec <peter.jausovec@solo.io> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: Peter Jausovec <peter.jausovec@solo.io>
1 parent 5be2fbc commit 7ad5567

61 files changed

Lines changed: 3516 additions & 641 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

.github/workflows/tag.yaml

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -104,8 +104,12 @@ jobs:
104104
export VERSION=$(echo "$GITHUB_REF" | cut -c12-)
105105
fi
106106
uv sync --all-extras
107+
uv version $VERSION --package kagent-core
108+
uv build --package kagent-core
107109
uv version $VERSION --package kagent-adk
108110
uv build --package kagent-adk
111+
uv version $VERSION --package kagent-langgraph
112+
uv build --package kagent-langgraph
109113
uv publish --token ${{ secrets.PYPI_TOKEN }}
110114
111115
release:

go/go.sum

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -128,10 +128,6 @@ github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8Hm
128128
github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
129129
github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
130130
github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
131-
github.com/kagent-dev/a2a-go v0.0.0-20250806145931-0fab01f644c3 h1:NrQCHyIOsQTtq9u7fpfQuDOMQAGVOa+n5Km/8csdqOM=
132-
github.com/kagent-dev/a2a-go v0.0.0-20250806145931-0fab01f644c3/go.mod h1:lu052zH/pTlTBwWMU/E3UckR0U9ajL1NlhiRkIF9R6A=
133-
github.com/kagent-dev/kmcp v0.1.4 h1:LozbTJYZCdPh8JDkFiCihevV4+l+eS1+cvmjD+1BpdI=
134-
github.com/kagent-dev/kmcp v0.1.4/go.mod h1:aPwM1QtoAackdbQqLrPaNSSg/KZa0fFjbTJyOZFJNK8=
135131
github.com/kagent-dev/kmcp v0.1.6 h1:O+W2g1tV2GcBHl04Tzln/wbP+VsVV6Mp/E1qCjUjYPQ=
136132
github.com/kagent-dev/kmcp v0.1.6/go.mod h1:aPwM1QtoAackdbQqLrPaNSSg/KZa0fFjbTJyOZFJNK8=
137133
github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=

go/internal/a2a/a2a_handler_mux.go

Lines changed: 6 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -3,9 +3,9 @@ package a2a
33
import (
44
"fmt"
55
"net/http"
6-
"strings"
76
"sync"
87

8+
"github.com/gorilla/mux"
99
authimpl "github.com/kagent-dev/kagent/go/internal/httpserver/auth"
1010
common "github.com/kagent-dev/kagent/go/internal/utils"
1111
"github.com/kagent-dev/kagent/go/pkg/auth"
@@ -78,15 +78,15 @@ func (a *handlerMux) getHandler(name string) (http.Handler, bool) {
7878

7979
func (a *handlerMux) ServeHTTP(w http.ResponseWriter, r *http.Request) {
8080

81+
vars := mux.Vars(r)
8182
// get the handler name from the first path segment
82-
path := strings.TrimPrefix(r.URL.Path, a.basePathPrefix)
83-
agentNamespace, remainingPath := popPath(path)
84-
if agentNamespace == "" {
83+
agentNamespace, ok := vars["namespace"]
84+
if !ok || agentNamespace == "" {
8585
http.Error(w, "Agent namespace not provided", http.StatusBadRequest)
8686
return
8787
}
88-
agentName, _ := popPath(remainingPath)
89-
if agentName == "" {
88+
agentName, ok := vars["name"]
89+
if !ok || agentName == "" {
9090
http.Error(w, "Agent name not provided", http.StatusBadRequest)
9191
return
9292
}
@@ -106,30 +106,3 @@ func (a *handlerMux) ServeHTTP(w http.ResponseWriter, r *http.Request) {
106106

107107
handlerHandler.ServeHTTP(w, r)
108108
}
109-
110-
// popPath separates the first element of a path from the rest.
111-
// It returns the first path element and the remaining path.
112-
// If the path is empty or only contains a separator, it returns empty strings.
113-
func popPath(path string) (firstElement, remainingPath string) {
114-
// Remove leading slash if present
115-
path = strings.TrimPrefix(path, "/")
116-
117-
// If path is empty after trimming, return empty strings
118-
if path == "" {
119-
return "", ""
120-
}
121-
122-
// Find the position of the first separator
123-
pos := strings.Index(path, "/")
124-
125-
// If no separator found, the first element is the entire path
126-
if pos == -1 {
127-
return path, ""
128-
}
129-
130-
// Split the path at the first separator
131-
firstElement = path[:pos]
132-
remainingPath = path[pos+1:] // Skip the separator
133-
134-
return firstElement, remainingPath
135-
}

go/internal/controller/translator/adk_api_translator.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1046,9 +1046,9 @@ func (a *adkApiTranslator) resolveByoDeployment(agent *v1alpha2.Agent) (*resolve
10461046
args = spec.Args
10471047
}
10481048

1049-
imagePullPolicy := corev1.PullIfNotPresent
1049+
imagePullPolicy := corev1.PullPolicy(DefaultImageConfig.PullPolicy)
10501050
if spec.ImagePullPolicy != "" {
1051-
imagePullPolicy = spec.ImagePullPolicy
1051+
imagePullPolicy = corev1.PullPolicy(spec.ImagePullPolicy)
10521052
}
10531053

10541054
replicas := spec.Replicas

go/internal/database/client.go

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,17 @@ type Client interface {
5151

5252
// Helper methods
5353
RefreshToolsForServer(serverName string, groupKind string, tools ...*v1alpha2.MCPTool) error
54+
55+
// LangGraph Checkpoint methods
56+
StoreCheckpoint(checkpoint *LangGraphCheckpoint) error
57+
StoreCheckpointWrites(writes []*LangGraphCheckpointWrite) error
58+
ListCheckpoints(userID, threadID, checkpointNS string, checkpointID *string, limit int) ([]*LangGraphCheckpointTuple, error)
59+
DeleteCheckpoint(userID, threadID string) error
60+
}
61+
62+
type LangGraphCheckpointTuple struct {
63+
Checkpoint *LangGraphCheckpoint
64+
Writes []*LangGraphCheckpointWrite
5465
}
5566

5667
type clientImpl struct {
@@ -481,3 +492,89 @@ func (c *clientImpl) ListPushNotifications(taskID string) ([]*protocol.TaskPushN
481492
func (c *clientImpl) DeletePushNotification(taskID string) error {
482493
return delete[PushNotification](c.db, Clause{Key: "task_id", Value: taskID})
483494
}
495+
496+
// StoreCheckpoint stores a LangGraph checkpoint and its writes atomically
497+
func (c *clientImpl) StoreCheckpoint(checkpoint *LangGraphCheckpoint) error {
498+
err := save(c.db, checkpoint)
499+
if err != nil {
500+
return fmt.Errorf("failed to store checkpoint: %w", err)
501+
}
502+
503+
return nil
504+
}
505+
506+
func (c *clientImpl) StoreCheckpointWrites(writes []*LangGraphCheckpointWrite) error {
507+
return c.db.Transaction(func(tx *gorm.DB) error {
508+
for _, write := range writes {
509+
if err := save(tx, write); err != nil {
510+
return fmt.Errorf("failed to store checkpoint write: %w", err)
511+
}
512+
}
513+
return nil
514+
})
515+
}
516+
517+
// ListCheckpoints lists checkpoints for a thread, optionally filtered by beforeCheckpointID
518+
func (c *clientImpl) ListCheckpoints(userID, threadID, checkpointNS string, checkpointID *string, limit int) ([]*LangGraphCheckpointTuple, error) {
519+
520+
var checkpointTuples []*LangGraphCheckpointTuple
521+
if err := c.db.Transaction(func(tx *gorm.DB) error {
522+
query := c.db.Where(
523+
"user_id = ? AND thread_id = ? AND checkpoint_ns = ?",
524+
userID, threadID, checkpointNS,
525+
)
526+
527+
if checkpointID != nil {
528+
query = query.Where("checkpoint_id = ?", *checkpointID)
529+
} else {
530+
query = query.Order("checkpoint_id DESC")
531+
}
532+
533+
// Apply limit
534+
if limit > 0 {
535+
query = query.Limit(limit)
536+
}
537+
538+
var checkpoints []LangGraphCheckpoint
539+
err := query.Find(&checkpoints).Error
540+
if err != nil {
541+
return fmt.Errorf("failed to list checkpoints: %w", err)
542+
}
543+
544+
for _, checkpoint := range checkpoints {
545+
var writes []*LangGraphCheckpointWrite
546+
if err := tx.Where(
547+
"user_id = ? AND thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
548+
userID, threadID, checkpointNS, checkpoint.CheckpointID,
549+
).Order("task_id, write_idx").Find(&writes).Error; err != nil {
550+
return fmt.Errorf("failed to get checkpoint writes: %w", err)
551+
}
552+
checkpointTuples = append(checkpointTuples, &LangGraphCheckpointTuple{
553+
Checkpoint: &checkpoint,
554+
Writes: writes,
555+
})
556+
}
557+
return nil
558+
}); err != nil {
559+
return nil, fmt.Errorf("failed to list checkpoints: %w", err)
560+
}
561+
return checkpointTuples, nil
562+
}
563+
564+
// DeleteCheckpoint deletes a checkpoint and its writes atomically
565+
func (c *clientImpl) DeleteCheckpoint(userID, threadID string) error {
566+
clauses := []Clause{
567+
{Key: "user_id", Value: userID},
568+
{Key: "thread_id", Value: threadID},
569+
}
570+
return c.db.Transaction(func(tx *gorm.DB) error {
571+
if err := delete[LangGraphCheckpoint](tx, clauses...); err != nil {
572+
return fmt.Errorf("failed to delete checkpoint: %w", err)
573+
}
574+
if err := delete[LangGraphCheckpointWrite](tx, clauses...); err != nil {
575+
return fmt.Errorf("failed to delete checkpoint writes: %w", err)
576+
}
577+
return nil
578+
})
579+
580+
}

0 commit comments

Comments
 (0)