External/roboco
1
0
Fork 0
mirror of https://github.com/rennf93/roboco.git synced 2026-08-03 07:23:24 +02:00
Code Issues Packages Projects Releases Wiki Activity

Revert "feat: add Kubernetes manifests and A2A protocol support"

Browse Source 
This reverts commit b943eb7ac1.
This commit is contained in:
Renn F
2025-12-28 06:03:54 +01:00
parent b943eb7ac1
commit b4d4fb089f
31 changed files with 245 additions and 3705 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/argocd/apps/roboco.yaml
-56
View File
@@ -1,56 +0,0 @@
# ArgoCD Application for RoboCo
# This defines how ArgoCD should deploy and manage RoboCo
apiVersion: argoproj.io/v1alpha1
kind: Application
metadata:
name: roboco
namespace: argocd
finalizers:
- resources-finalizer.argocd.argoproj.io
spec:
project: default
source:
# GitHub repository containing K8s manifests
repoURL: git@github.com:renzof/roboco.git
targetRevision: main
path: deploy/overlays/production
destination:
# Deploy to local cluster
server: https://kubernetes.default.svc
namespace: roboco
syncPolicy:
automated:
# Automatically prune resources that are no longer in git
prune: true
# Automatically sync when ArgoCD detects drift
selfHeal: true
# Don't allow empty directories
allowEmpty: false
syncOptions:
# Create namespace if it doesn't exist
- CreateNamespace=true
# Use foreground deletion for proper cleanup
- PrunePropagationPolicy=foreground
retry:
limit: 5
backoff:
duration: 5s
factor: 2
maxDuration: 3m
# Ignore differences that are expected (e.g., HPA manages replicas)
ignoreDifferences:
- group: apps
kind: Deployment
jsonPointers:
- /spec/replicas
# Health checks
info:
- name: description
value: "RoboCo AI Agentic Company"
- name: maintainer
value: "renzof"
deploy/base/api/deployment.yaml
-83
View File
@@ -1,83 +0,0 @@
# RoboCo API Deployment
apiVersion: apps/v1
kind: Deployment
metadata:
name: roboco-api
namespace: roboco
labels:
app.kubernetes.io/name: roboco-api
app.kubernetes.io/component: api
spec:
replicas: 2
selector:
matchLabels:
app: roboco-api
template:
metadata:
labels:
app: roboco-api
app.kubernetes.io/name: roboco-api
spec:
containers:
- name: api
image: ghcr.io/renzof/roboco-api:latest
ports:
- containerPort: 8000
name: http
env:
# Database
- name: ROBOCO_DATABASE_HOST
value: postgres
- name: ROBOCO_DATABASE_PORT
value: "5432"
- name: ROBOCO_DATABASE_USER
valueFrom:
secretKeyRef:
name: roboco-secrets
key: postgres-user
- name: ROBOCO_DATABASE_PASSWORD
valueFrom:
secretKeyRef:
name: roboco-secrets
key: postgres-password
- name: ROBOCO_DATABASE_NAME
value: roboco
# Redis
- name: ROBOCO_REDIS_HOST
value: redis
- name: ROBOCO_REDIS_PORT
value: "6379"
# API
- name: ROBOCO_HOST
value: "0.0.0.0"
- name: ROBOCO_PORT
value: "8000"
- name: ROBOCO_ENVIRONMENT
value: production
# LLM
- name: ROBOCO_ANTHROPIC_API_KEY
valueFrom:
secretKeyRef:
name: roboco-secrets
key: anthropic-api-key
resources:
requests:
memory: "256Mi"
cpu: "100m"
limits:
memory: "1Gi"
cpu: "500m"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 15
periodSeconds: 10
timeoutSeconds: 5
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 5
periodSeconds: 5
timeoutSeconds: 3
deploy/base/api/service.yaml
-36
View File
@@ -1,36 +0,0 @@
# RoboCo API Service
apiVersion: v1
kind: Service
metadata:
name: roboco-api
namespace: roboco
labels:
app.kubernetes.io/name: roboco-api
app.kubernetes.io/component: api
spec:
type: ClusterIP
ports:
- port: 8000
targetPort: 8000
name: http
selector:
app: roboco-api
---
# NodePort for external access (can be replaced with Ingress)
apiVersion: v1
kind: Service
metadata:
name: roboco-api-external
namespace: roboco
labels:
app.kubernetes.io/name: roboco-api
app.kubernetes.io/component: api
spec:
type: NodePort
ports:
- port: 8000
targetPort: 8000
nodePort: 30080
name: http
selector:
app: roboco-api
deploy/base/kustomization.yaml
-41
View File
@@ -1,41 +0,0 @@
# Base Kustomization for RoboCo
apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
namespace: roboco
resources:
# Namespace
- namespace/namespace.yaml
# Storage
- storage/storageclass.yaml
# PostgreSQL
- postgres/statefulset.yaml
- postgres/service.yaml
# Redis
- redis/deployment.yaml
- redis/service.yaml
- redis/pvc.yaml
# API
- api/deployment.yaml
- api/service.yaml
# Orchestrator
- orchestrator/serviceaccount.yaml
- orchestrator/role.yaml
- orchestrator/rolebinding.yaml
- orchestrator/deployment.yaml
- orchestrator/service.yaml
# Common labels applied to all resources
commonLabels:
app.kubernetes.io/part-of: roboco
app.kubernetes.io/managed-by: kustomize
# Image configurations (can be overridden in overlays)
images:
- name: ghcr.io/renzof/roboco-api
newTag: latest
- name: ghcr.io/renzof/roboco-orchestrator
newTag: latest
- name: ghcr.io/renzof/roboco-agent
newTag: latest
deploy/base/namespace/namespace.yaml
-9
View File
@@ -1,9 +0,0 @@
# RoboCo Namespace
# All RoboCo resources are deployed to this namespace
apiVersion: v1
kind: Namespace
metadata:
name: roboco
labels:
app.kubernetes.io/name: roboco
app.kubernetes.io/component: namespace
deploy/base/orchestrator/deployment.yaml
-110
View File
@@ -1,110 +0,0 @@
# RoboCo Orchestrator Deployment
# Manages agent lifecycle and task dispatching
apiVersion: apps/v1
kind: Deployment
metadata:
name: roboco-orchestrator
namespace: roboco
labels:
app.kubernetes.io/name: roboco-orchestrator
app.kubernetes.io/component: orchestrator
spec:
replicas: 1 # Single instance for now (state coordination needed for HA)
selector:
matchLabels:
app: roboco-orchestrator
template:
metadata:
labels:
app: roboco-orchestrator
app.kubernetes.io/name: roboco-orchestrator
spec:
serviceAccountName: roboco-orchestrator
containers:
- name: orchestrator
image: ghcr.io/renzof/roboco-orchestrator:latest
ports:
- containerPort: 8000
name: http
env:
# Database
- name: ROBOCO_DATABASE_HOST
value: postgres
- name: ROBOCO_DATABASE_PORT
value: "5432"
- name: ROBOCO_DATABASE_USER
valueFrom:
secretKeyRef:
name: roboco-secrets
key: postgres-user
- name: ROBOCO_DATABASE_PASSWORD
valueFrom:
secretKeyRef:
name: roboco-secrets
key: postgres-password
- name: ROBOCO_DATABASE_NAME
value: roboco
# Redis
- name: ROBOCO_REDIS_HOST
value: redis
- name: ROBOCO_REDIS_PORT
value: "6379"
# API
- name: ROBOCO_HOST
value: "0.0.0.0"
- name: ROBOCO_PORT
value: "8000"
- name: ROBOCO_API_URL
value: "http://roboco-api:8000"
- name: ROBOCO_ENVIRONMENT
value: production
# K8s settings
- name: ROBOCO_K8S_ENABLED
value: "true"
- name: ROBOCO_K8S_NAMESPACE
valueFrom:
fieldRef:
fieldPath: metadata.namespace
- name: ROBOCO_K8S_AGENT_IMAGE
value: "ghcr.io/renzof/roboco-agent"
# LLM
- name: ROBOCO_ANTHROPIC_API_KEY
valueFrom:
secretKeyRef:
name: roboco-secrets
key: anthropic-api-key
volumeMounts:
- name: blueprints
mountPath: /app/agents/blueprints
readOnly: true
- name: claude-auth
mountPath: /root/.claude
resources:
requests:
memory: "256Mi"
cpu: "100m"
limits:
memory: "1Gi"
cpu: "500m"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 15
periodSeconds: 10
timeoutSeconds: 5
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 5
periodSeconds: 5
timeoutSeconds: 3
volumes:
- name: blueprints
configMap:
name: roboco-blueprints
- name: claude-auth
secret:
secretName: claude-auth
optional: true
deploy/base/orchestrator/role.yaml
-31
View File
@@ -1,31 +0,0 @@
# Orchestrator Role
# Permissions for spawning and managing agent Jobs
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
name: roboco-orchestrator
namespace: roboco
labels:
app.kubernetes.io/name: roboco-orchestrator
app.kubernetes.io/component: orchestrator
rules:
# Create/manage agent Jobs
- apiGroups: ["batch"]
resources: ["jobs"]
verbs: ["create", "delete", "get", "list", "watch", "patch"]
# Watch pods for job status
- apiGroups: [""]
resources: ["pods"]
verbs: ["get", "list", "watch"]
# Read pod logs for debugging
- apiGroups: [""]
resources: ["pods/log"]
verbs: ["get"]
# Manage ConfigMaps for agent configs (MCP configs, prompts)
- apiGroups: [""]
resources: ["configmaps"]
verbs: ["create", "delete", "get", "list", "update", "patch"]
# Read secrets for agent environment (API keys)
- apiGroups: [""]
resources: ["secrets"]
verbs: ["get"]
deploy/base/orchestrator/rolebinding.yaml
-17
View File
@@ -1,17 +0,0 @@
# Orchestrator RoleBinding
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
name: roboco-orchestrator
namespace: roboco
labels:
app.kubernetes.io/name: roboco-orchestrator
app.kubernetes.io/component: orchestrator
subjects:
- kind: ServiceAccount
name: roboco-orchestrator
namespace: roboco
roleRef:
kind: Role
name: roboco-orchestrator
apiGroup: rbac.authorization.k8s.io
deploy/base/orchestrator/service.yaml
-17
View File
@@ -1,17 +0,0 @@
# Orchestrator Service (internal only)
apiVersion: v1
kind: Service
metadata:
name: roboco-orchestrator
namespace: roboco
labels:
app.kubernetes.io/name: roboco-orchestrator
app.kubernetes.io/component: orchestrator
spec:
type: ClusterIP
ports:
- port: 8000
targetPort: 8000
name: http
selector:
app: roboco-orchestrator
deploy/base/orchestrator/serviceaccount.yaml
-10
View File
@@ -1,10 +0,0 @@
# Orchestrator ServiceAccount
# Used by the orchestrator to interact with K8s API for spawning agent Jobs
apiVersion: v1
kind: ServiceAccount
metadata:
name: roboco-orchestrator
namespace: roboco
labels:
app.kubernetes.io/name: roboco-orchestrator
app.kubernetes.io/component: orchestrator
deploy/base/postgres/service.yaml
-17
View File
@@ -1,17 +0,0 @@
# PostgreSQL Service
apiVersion: v1
kind: Service
metadata:
name: postgres
namespace: roboco
labels:
app.kubernetes.io/name: postgres
app.kubernetes.io/component: database
spec:
type: ClusterIP
ports:
- port: 5432
targetPort: 5432
name: postgres
selector:
app: postgres
deploy/base/postgres/statefulset.yaml
-84
View File
@@ -1,84 +0,0 @@
# PostgreSQL StatefulSet with pgvector for RAG
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: postgres
namespace: roboco
labels:
app.kubernetes.io/name: postgres
app.kubernetes.io/component: database
spec:
serviceName: postgres
replicas: 1
selector:
matchLabels:
app: postgres
template:
metadata:
labels:
app: postgres
app.kubernetes.io/name: postgres
spec:
containers:
- name: postgres
image: pgvector/pgvector:pg16
ports:
- containerPort: 5432
name: postgres
env:
- name: POSTGRES_USER
valueFrom:
secretKeyRef:
name: roboco-secrets
key: postgres-user
- name: POSTGRES_PASSWORD
valueFrom:
secretKeyRef:
name: roboco-secrets
key: postgres-password
- name: POSTGRES_DB
value: roboco
- name: PGDATA
value: /var/lib/postgresql/data/pgdata
volumeMounts:
- name: postgres-data
mountPath: /var/lib/postgresql/data
resources:
requests:
memory: "512Mi"
cpu: "250m"
limits:
memory: "2Gi"
cpu: "1000m"
livenessProbe:
exec:
command:
- pg_isready
- -U
- roboco
- -d
- roboco
initialDelaySeconds: 30
periodSeconds: 10
timeoutSeconds: 5
failureThreshold: 6
readinessProbe:
exec:
command:
- pg_isready
- -U
- roboco
- -d
- roboco
initialDelaySeconds: 5
periodSeconds: 5
timeoutSeconds: 3
volumeClaimTemplates:
- metadata:
name: postgres-data
spec:
accessModes: ["ReadWriteOnce"]
storageClassName: nfs-roboco
resources:
requests:
storage: 10Gi
deploy/base/redis/deployment.yaml
-64
View File
@@ -1,64 +0,0 @@
# Redis Deployment for cache, sessions, and event bus
apiVersion: apps/v1
kind: Deployment
metadata:
name: redis
namespace: roboco
labels:
app.kubernetes.io/name: redis
app.kubernetes.io/component: cache
spec:
replicas: 1
selector:
matchLabels:
app: redis
template:
metadata:
labels:
app: redis
app.kubernetes.io/name: redis
spec:
containers:
- name: redis
image: redis:8-alpine
command:
- redis-server
- --appendonly
- "yes"
- --maxmemory
- "256mb"
- --maxmemory-policy
- allkeys-lru
ports:
- containerPort: 6379
name: redis
volumeMounts:
- name: redis-data
mountPath: /data
resources:
requests:
memory: "128Mi"
cpu: "100m"
limits:
memory: "512Mi"
cpu: "500m"
livenessProbe:
exec:
command:
- redis-cli
- ping
initialDelaySeconds: 10
periodSeconds: 10
timeoutSeconds: 5
readinessProbe:
exec:
command:
- redis-cli
- ping
initialDelaySeconds: 5
periodSeconds: 5
timeoutSeconds: 3
volumes:
- name: redis-data
persistentVolumeClaim:
claimName: redis-data
deploy/base/redis/pvc.yaml
-16
View File
@@ -1,16 +0,0 @@
# Redis Persistent Volume Claim
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: redis-data
namespace: roboco
labels:
app.kubernetes.io/name: redis
app.kubernetes.io/component: cache
spec:
accessModes:
- ReadWriteOnce
storageClassName: nfs-roboco
resources:
requests:
storage: 1Gi
deploy/base/redis/service.yaml
-17
View File
@@ -1,17 +0,0 @@
# Redis Service
apiVersion: v1
kind: Service
metadata:
name: redis
namespace: roboco
labels:
app.kubernetes.io/name: redis
app.kubernetes.io/component: cache
spec:
type: ClusterIP
ports:
- port: 6379
targetPort: 6379
name: redis
selector:
app: redis
deploy/base/storage/storageclass.yaml
-19
View File
@@ -1,19 +0,0 @@
# NFS StorageClass for RoboCo
# Uses NFS CSI driver for persistent storage on UGREEN NAS
apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
name: nfs-roboco
provisioner: nfs.csi.k8s.io
parameters:
# NFS server address (UGREEN NAS)
server: 192.168.50.111
# NFS share path
share: /volume1/roboco/k8s-data
reclaimPolicy: Retain
volumeBindingMode: Immediate
mountOptions:
- nfsvers=4.1
- hard
- timeo=600
- retrans=2
deploy/overlays/development/kustomization.yaml
-56
View File
@@ -1,56 +0,0 @@
# Development Overlay for RoboCo
apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
namespace: roboco
resources:
- ../../base
# Development-specific patches
patches:
# Single replica for development
- patch: |-
- op: replace
path: /spec/replicas
value: 1
target:
kind: Deployment
name: roboco-api
# Lower resource limits for development
- patch: |-
- op: replace
path: /spec/template/spec/containers/0/resources/limits/memory
value: "512Mi"
- op: replace
path: /spec/template/spec/containers/0/resources/limits/cpu
value: "250m"
target:
kind: Deployment
name: roboco-api
# Development PostgreSQL resources
- patch: |-
- op: replace
path: /spec/template/spec/containers/0/resources/limits/memory
value: "1Gi"
- op: replace
path: /spec/template/spec/containers/0/resources/limits/cpu
value: "500m"
target:
kind: StatefulSet
name: postgres
# Image tags for development (use latest)
images:
- name: ghcr.io/renzof/roboco-api
newTag: latest
- name: ghcr.io/renzof/roboco-orchestrator
newTag: latest
# Config overrides
configMapGenerator:
- name: roboco-config
literals:
- ENVIRONMENT=development
- LOG_LEVEL=DEBUG
- DEBUG=true
deploy/overlays/production/kustomization.yaml
-58
View File
@@ -1,58 +0,0 @@
# Production Overlay for RoboCo
apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
namespace: roboco
resources:
- ../../base
# Sealed secrets for production
# - sealed-secrets.yaml
# Production-specific patches
patches:
# Increase API replicas
- patch: |-
- op: replace
path: /spec/replicas
value: 3
target:
kind: Deployment
name: roboco-api
# Increase resource limits for production
- patch: |-
- op: replace
path: /spec/template/spec/containers/0/resources/limits/memory
value: "2Gi"
- op: replace
path: /spec/template/spec/containers/0/resources/limits/cpu
value: "1000m"
target:
kind: Deployment
name: roboco-api
# Production PostgreSQL resources
- patch: |-
- op: replace
path: /spec/template/spec/containers/0/resources/limits/memory
value: "4Gi"
- op: replace
path: /spec/template/spec/containers/0/resources/limits/cpu
value: "2000m"
target:
kind: StatefulSet
name: postgres
# Image tags for production
images:
- name: ghcr.io/renzof/roboco-api
newTag: stable
- name: ghcr.io/renzof/roboco-orchestrator
newTag: stable
# Config overrides
configMapGenerator:
- name: roboco-config
literals:
- ENVIRONMENT=production
- LOG_LEVEL=INFO
- DEBUG=false
pyproject.toml
-7
View File
@@ -48,10 +48,6 @@ dependencies = [
"passlib[bcrypt]", # Password hashing
"tenacity", # Retry logic
"structlog", # Structured logging
# Container Orchestration
"kubernetes", # K8s API client for agent Jobs
"sse-starlette", # Server-Sent Events for A2A streaming
]
[project.optional-dependencies]
@@ -142,7 +138,6 @@ select = [
"roboco/mcp/**/*.py" = ["PLC0415"]
"roboco/services/*.py" = ["PLC0415"]
"roboco/api/routes/*.py" = ["PLC0415"]
"roboco/runtime/*.py" = ["PLC0415"]
# =============================================================================
# MyPy Configuration
@@ -172,8 +167,6 @@ module = [
"tiktoken.*",
"piragi.*",
"toon.*",
"kubernetes.*",
"sse_starlette.*",
]
ignore_missing_imports = true
roboco/api/app.py
-13
View File
@@ -12,8 +12,6 @@ from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from roboco.api.middleware import setup_middleware
from roboco.api.routes.a2a import router as a2a_router
from roboco.api.routes.a2a import wellknown_router as a2a_wellknown_router
from roboco.api.routes.agents import router as agents_router
from roboco.api.routes.channels import router as channels_router
from roboco.api.routes.dashboard import router as dashboard_router
@@ -148,10 +146,6 @@ def create_app() -> FastAPI:
# Health check
app.include_router(health_router, tags=["Health"])
# A2A Protocol: Well-known endpoints at root level
# (/.well-known/agent.json, /agents/{id}/.well-known/agent.json)
app.include_router(a2a_wellknown_router, tags=["A2A Protocol"])
# API v1
api_prefix = "/api/v1"
@@ -237,13 +231,6 @@ def create_app() -> FastAPI:
tags=["Orchestrator"],
)
# A2A Protocol: API endpoints
app.include_router(
a2a_router,
prefix=f"{api_prefix}/a2a",
tags=["A2A Protocol"],
)
# ==========================================================================
# WebSocket
# ==========================================================================
roboco/api/routes/a2a.py
-816
View File
@@ -1,816 +0,0 @@
"""
A2A (Agent-to-Agent) Protocol Routes
Implements Google's A2A protocol for agent interoperability.
See: https://a2a-protocol.org/latest/specification/
Endpoints:
- GET /.well-known/agent.json: System Agent Card
- GET /agents/{agent_id}/.well-known/agent.json: Per-agent Agent Card
- POST /api/v1/a2a/message/send: Send message and create/update task
- POST /api/v1/a2a/message/stream: Send message with SSE streaming
- GET /api/v1/a2a/tasks/{task_id}: Get task state
- GET /api/v1/a2a/tasks: List tasks
- POST /api/v1/a2a/tasks/{task_id}/cancel: Cancel task
"""
import asyncio
import contextlib
from collections.abc import AsyncGenerator
from typing import Any
from uuid import UUID
from fastapi import APIRouter, HTTPException, Query, Request, status
from fastapi.responses import JSONResponse
from sqlalchemy import select
from sse_starlette import EventSourceResponse
from roboco.api.deps import DbSession
from roboco.config import settings
from roboco.db.tables import AgentTable, TaskTable
from roboco.models.a2a import (
A2AArtifact,
A2AMessage,
A2ATask,
A2ATaskStatus,
AgentCapabilities,
AgentCard,
AgentProvider,
AgentSkill,
CancelTaskRequest,
ListTasksResponse,
SecurityScheme,
SendMessageRequest,
SendMessageResponse,
TextPart,
task_status_to_a2a_state,
)
from roboco.models.base import TaskStatus
# Router for A2A API endpoints (mounted at /api/v1/a2a)
router = APIRouter()
# Router for well-known endpoints (mounted at root level)
wellknown_router = APIRouter()
# =============================================================================
# HELPER FUNCTIONS
# =============================================================================
def _get_service_endpoint() -> str:
"""Build service endpoint URL from settings."""
connect_host = "127.0.0.1" if settings.host == "0.0.0.0" else settings.host
return f"http://{connect_host}:{settings.port}"
def _build_system_agent_card() -> AgentCard:
"""Build the system-level Agent Card for RoboCo."""
return AgentCard(
id="roboco-system",
name="RoboCo System",
description=(
"RoboCo is an AI Agentic Company - a virtual organization of AI agents "
"designed to operate as a complete software development workforce."
),
provider=AgentProvider(
organization="RoboCo",
url="https://github.com/roboco",
),
protocol_version="1.0",
service_endpoint=f"{_get_service_endpoint()}/api/v1/a2a",
version=settings.app_version,
capabilities=AgentCapabilities(
streaming=True, # We support SSE
push_notifications=False, # Not implemented yet
state_transition_history=True, # We track task history
),
default_input_modes=["text/plain", "application/json"],
default_output_modes=["text/plain", "application/json"],
skills=[
AgentSkill(
id="software-development",
name="Software Development",
description="Full-stack software development with AI agents",
tags=["development", "coding", "qa", "documentation"],
),
AgentSkill(
id="task-management",
name="Task Management",
description="Create and manage development tasks",
tags=["tasks", "kanban", "planning"],
),
AgentSkill(
id="code-review",
name="Code Review",
description="Review and quality assurance of code",
tags=["qa", "review", "testing"],
),
],
documentation_url="https://github.com/roboco/docs",
security_schemes={
"bearerAuth": SecurityScheme(type="http", scheme="bearer"),
},
security=[{"bearerAuth": []}],
)
async def _build_agent_card(agent: AgentTable) -> AgentCard:
"""Build an Agent Card for a specific agent."""
agent_id = str(agent.id)
agent_slug = agent.slug
# Map role to skills
role_skills: dict[str, list[AgentSkill]] = {
"developer": [
AgentSkill(
id="coding",
name="Code Development",
description="Write and implement code",
tags=["development", "coding"],
),
AgentSkill(
id="debugging",
name="Debugging",
description="Debug and fix code issues",
tags=["debugging", "troubleshooting"],
),
],
"qa": [
AgentSkill(
id="testing",
name="Testing",
description="Test code and verify quality",
tags=["qa", "testing"],
),
AgentSkill(
id="review",
name="Code Review",
description="Review code for quality and issues",
tags=["qa", "review"],
),
],
"documenter": [
AgentSkill(
id="documentation",
name="Documentation",
description="Write technical documentation",
tags=["documentation", "writing"],
),
],
"cell_pm": [
AgentSkill(
id="coordination",
name="Task Coordination",
description="Coordinate tasks within the cell",
tags=["management", "coordination"],
),
],
"main_pm": [
AgentSkill(
id="planning",
name="Project Planning",
description="Plan and coordinate across cells",
tags=["management", "planning"],
),
],
}
skills = role_skills.get(agent.role, [])
return AgentCard(
id=agent_id,
name=agent.name,
description=f"{agent.name} - {agent.role} agent in RoboCo",
provider=AgentProvider(
organization="RoboCo",
url="https://github.com/roboco",
),
protocol_version="1.0",
service_endpoint=f"{_get_service_endpoint()}/api/v1/a2a",
version=settings.app_version,
capabilities=AgentCapabilities(
streaming=True,
push_notifications=False,
state_transition_history=True,
),
default_input_modes=["text/plain", "application/json"],
default_output_modes=["text/plain", "application/json"],
skills=skills,
metadata={
"slug": agent_slug,
"role": agent.role,
"team": agent.team,
},
security_schemes={
"bearerAuth": SecurityScheme(type="http", scheme="bearer"),
},
security=[{"bearerAuth": []}],
)
def _task_to_a2a(task: TaskTable) -> A2ATask:
"""Convert a RoboCo TaskTable to A2A Task."""
task_id = str(task.id)
# Build status - get status value as string
if hasattr(task.status, "value"):
status_value = task.status.value
else:
status_value = str(task.status)
a2a_state = task_status_to_a2a_state(status_value)
status_message = None
if task.dev_notes:
status_message = A2AMessage(
role="agent",
parts=[TextPart(text=task.dev_notes)],
task_id=task_id,
)
a2a_status = A2ATaskStatus(
state=a2a_state,
message=status_message,
timestamp=task.updated_at or task.created_at,
)
# Build artifacts from task outputs (if any)
artifacts: list[A2AArtifact] = []
# Build metadata from task fields
metadata: dict[str, Any] = {
"roboco_status": status_value,
"priority": task.priority,
"team": task.team,
}
if task.assigned_to:
metadata["assigned_to"] = str(task.assigned_to)
if task.parent_task_id:
metadata["parent_task_id"] = str(task.parent_task_id)
return A2ATask(
id=task_id,
context_id=task_id, # Use task_id as context_id
status=a2a_status,
artifacts=artifacts,
history=[], # Would need to load from message history
metadata=metadata,
)
# =============================================================================
# WELL-KNOWN ENDPOINTS (mounted at root)
# =============================================================================
@wellknown_router.get("/.well-known/agent.json")
async def get_system_agent_card() -> JSONResponse:
"""
Get the system-level Agent Card.
Per A2A specification, returns the agent's public identity and capabilities.
"""
card = _build_system_agent_card()
return JSONResponse(
content=card.model_dump(by_alias=True, exclude_none=True),
media_type="application/json",
)
@wellknown_router.get("/agents/{agent_id}/.well-known/agent.json")
async def get_agent_card(
agent_id: str,
db: DbSession,
) -> JSONResponse:
"""
Get Agent Card for a specific agent.
Accepts either a UUID string or agent slug (e.g., "be-dev-1").
"""
# Try to parse as UUID first
try:
uuid = UUID(agent_id)
result = await db.execute(select(AgentTable).where(AgentTable.id == uuid))
except ValueError:
# Not a UUID, try slug lookup
result = await db.execute(
select(AgentTable).where(AgentTable.slug == agent_id)
)
agent = result.scalar_one_or_none()
if agent is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Agent not found: {agent_id}",
)
card = await _build_agent_card(agent)
return JSONResponse(
content=card.model_dump(by_alias=True, exclude_none=True),
media_type="application/json",
)
# =============================================================================
# A2A API ENDPOINTS (mounted at /api/v1/a2a)
# =============================================================================
@router.post("/message/send")
async def send_message(
request: SendMessageRequest,
db: DbSession,
) -> SendMessageResponse:
"""
Send a message to create or update an A2A task.
This is the primary A2A interaction endpoint. Messages sent here
create new tasks or continue existing conversations.
"""
message = request.message
# Extract task_id from message if present
task_id_str = message.task_id
if task_id_str:
# Update existing task
try:
task_uuid = UUID(task_id_str)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid task ID: {task_id_str}",
) from None
result = await db.execute(
select(TaskTable).where(TaskTable.id == task_uuid)
)
task = result.scalar_one_or_none()
if task is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Task not found: {task_id_str}",
)
# Update task dev_notes with new message
text_parts = [p for p in message.parts if p.type == "text"]
if text_parts:
text_part = text_parts[0]
if hasattr(text_part, "text"):
new_text = text_part.text
if task.dev_notes:
task.dev_notes = f"{task.dev_notes}\n\n{new_text}"
else:
task.dev_notes = new_text
await db.commit()
await db.refresh(task)
else:
# Create new task from message
# Note: For proper implementation, this should use TaskService
# which handles session creation and proper agent context
text_parts = [p for p in message.parts if p.type == "text"]
title = "A2A Task"
description = ""
if text_parts:
text_part = text_parts[0]
if hasattr(text_part, "text"):
first_text = text_part.text
# Use first line as title, rest as description
lines = first_text.split("\n", 1)
title = lines[0][:200] # Truncate title
description = lines[1] if len(lines) > 1 else first_text
# Get a system agent to use as creator
# In production, this should come from authenticated context
result = await db.execute(
select(AgentTable).where(AgentTable.role == "main_pm").limit(1)
)
system_agent = result.scalar_one_or_none()
if system_agent is None:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="No system agent available to create tasks",
)
from roboco.models.base import Team
task = TaskTable(
title=title,
description=description,
acceptance_criteria=["Task completed as specified"],
status=TaskStatus.PENDING,
priority=5,
team=Team.BACKEND,
created_by=system_agent.id,
)
db.add(task)
await db.commit()
await db.refresh(task)
return SendMessageResponse(task=_task_to_a2a(task))
@router.post("/message/stream")
async def send_message_stream(
request: Request,
body: SendMessageRequest,
db: DbSession,
) -> EventSourceResponse:
"""
Send a message with SSE streaming response.
Per A2A specification, this endpoint streams task updates in real-time
as the task progresses through its lifecycle.
Returns Server-Sent Events with task state updates.
"""
message = body.message
async def generate_task_events() -> AsyncGenerator[dict[str, Any]]:
"""Generate SSE events for task lifecycle."""
# Create or get task
task_id_str = message.task_id
if task_id_str:
# Get existing task
try:
task_uuid = UUID(task_id_str)
except ValueError:
yield {
"event": "error",
"data": f"Invalid task ID: {task_id_str}",
}
return
result = await db.execute(
select(TaskTable).where(TaskTable.id == task_uuid)
)
task = result.scalar_one_or_none()
if task is None:
yield {
"event": "error",
"data": f"Task not found: {task_id_str}",
}
return
# Send initial task state
a2a_task = _task_to_a2a(task)
yield {
"event": "task.status",
"id": str(task.id),
"data": a2a_task.model_dump_json(by_alias=True),
}
# Stream updates while task is in progress
poll_count = 0
max_polls = 60 # Poll for up to 60 iterations (5 minutes at 5s interval)
while poll_count < max_polls:
# Check for client disconnect
if await request.is_disconnected():
break
await asyncio.sleep(5) # Poll interval
poll_count += 1
# Refresh task state
await db.refresh(task)
# Get current status
if hasattr(task.status, "value"):
current_status = task.status.value
else:
current_status = str(task.status)
# Send update
a2a_task = _task_to_a2a(task)
yield {
"event": "task.status",
"id": f"{task.id}-{poll_count}",
"data": a2a_task.model_dump_json(by_alias=True),
}
# Stop if task is in terminal state
if current_status in ["completed", "cancelled"]:
yield {
"event": "task.complete",
"id": f"{task.id}-final",
"data": a2a_task.model_dump_json(by_alias=True),
}
break
else:
# New task - send creation event
yield {
"event": "task.creating",
"data": "Creating new task from message...",
}
# Note: Full task creation logic would go here
# For now, send a placeholder
yield {
"event": "error",
"data": "Task creation via streaming not yet implemented",
}
return EventSourceResponse(
generate_task_events(),
ping=15, # Keep connection alive every 15 seconds
)
@router.get("/tasks/{task_id}/subscribe")
async def subscribe_to_task(
request: Request,
task_id: str,
db: DbSession,
) -> EventSourceResponse:
"""
Subscribe to task updates via SSE.
Opens a persistent connection that streams task state changes
until the task reaches a terminal state or client disconnects.
"""
try:
task_uuid = UUID(task_id)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid task ID: {task_id}",
) from None
result = await db.execute(select(TaskTable).where(TaskTable.id == task_uuid))
task = result.scalar_one_or_none()
if task is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Task not found: {task_id}",
)
async def generate_updates() -> AsyncGenerator[dict[str, Any]]:
"""Stream task updates."""
poll_count = 0
max_polls = 720 # 1 hour at 5s interval
last_status = None
while poll_count < max_polls:
if await request.is_disconnected():
break
# Refresh task state from DB
await db.refresh(task)
# Get current status
if hasattr(task.status, "value"):
current_status = task.status.value
else:
current_status = str(task.status)
# Only send update if status changed
if current_status != last_status:
a2a_task = _task_to_a2a(task)
yield {
"event": "task.status",
"id": f"{task_id}-{poll_count}",
"data": a2a_task.model_dump_json(by_alias=True),
}
last_status = current_status
# Stop if terminal
if current_status in ["completed", "cancelled"]:
yield {
"event": "task.complete",
"id": f"{task_id}-final",
"data": a2a_task.model_dump_json(by_alias=True),
}
break
await asyncio.sleep(5)
poll_count += 1
return EventSourceResponse(
generate_updates(),
ping=15,
)
@router.get("/tasks/{task_id}")
async def get_task(
task_id: str,
db: DbSession,
_history_length: int | None = Query(
None, alias="historyLength", description="Number of history turns to include"
),
) -> A2ATask:
"""
Get the state of an A2A task.
Returns task details including status, artifacts, and optionally history.
"""
try:
task_uuid = UUID(task_id)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid task ID: {task_id}",
) from None
result = await db.execute(select(TaskTable).where(TaskTable.id == task_uuid))
task = result.scalar_one_or_none()
if task is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Task not found: {task_id}",
)
return _task_to_a2a(task)
@router.get("/tasks")
async def list_tasks(
db: DbSession,
page_size: int = Query(20, alias="pageSize", ge=1, le=100),
page_token: str | None = Query(None, alias="pageToken"),
_filter_str: str | None = Query(None, alias="filter"),
order_by: str | None = Query(None, alias="orderBy"),
) -> ListTasksResponse:
"""
List A2A tasks with pagination.
Supports filtering and pagination via page tokens.
"""
query = select(TaskTable)
# Apply ordering
if order_by:
if order_by == "created_at desc":
query = query.order_by(TaskTable.created_at.desc())
elif order_by == "created_at asc":
query = query.order_by(TaskTable.created_at.asc())
else:
query = query.order_by(TaskTable.created_at.desc())
else:
query = query.order_by(TaskTable.created_at.desc())
# Handle pagination
offset = 0
if page_token:
with contextlib.suppress(ValueError):
offset = int(page_token)
query = query.offset(offset).limit(page_size + 1)
result = await db.execute(query)
tasks = list(result.scalars().all())
# Check if there are more results
has_more = len(tasks) > page_size
if has_more:
tasks = tasks[:page_size]
next_page_token = str(offset + page_size) if has_more else None
return ListTasksResponse(
tasks=[_task_to_a2a(t) for t in tasks],
next_page_token=next_page_token,
)
@router.post("/tasks/{task_id}/cancel")
async def cancel_task(
task_id: str,
db: DbSession,
request: CancelTaskRequest | None = None,
) -> A2ATask:
"""
Cancel an A2A task.
Transitions the task to cancelled state.
"""
try:
task_uuid = UUID(task_id)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid task ID: {task_id}",
) from None
result = await db.execute(select(TaskTable).where(TaskTable.id == task_uuid))
task = result.scalar_one_or_none()
if task is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Task not found: {task_id}",
)
# Check if task can be cancelled
if hasattr(task.status, "value"):
status_value = task.status.value
else:
status_value = str(task.status)
terminal_states = ["completed", "cancelled"]
if status_value in terminal_states:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Task already in terminal state: {status_value}",
)
# Cancel the task
task.status = TaskStatus.CANCELLED
if request and request.reason:
reason_text = f"Cancellation reason: {request.reason}"
if task.dev_notes:
task.dev_notes = f"{task.dev_notes}\n\n{reason_text}"
else:
task.dev_notes = reason_text
await db.commit()
await db.refresh(task)
return _task_to_a2a(task)
# =============================================================================
# AGENT DISCOVERY ENDPOINTS
# =============================================================================
@router.get("/agents")
async def discover_agents(
db: DbSession,
role: str | None = Query(None, description="Filter by agent role"),
team: str | None = Query(None, description="Filter by team"),
skill: str | None = Query(None, description="Filter by skill tag"),
) -> list[AgentCard]:
"""
Discover agents matching criteria.
Returns a list of AgentCards for agents that match the specified filters.
This enables A2A clients to find agents with specific capabilities.
"""
query = select(AgentTable)
if role:
query = query.where(AgentTable.role == role)
if team:
query = query.where(AgentTable.team == team)
result = await db.execute(query)
agents = result.scalars().all()
# Build cards for all matching agents
cards = []
for agent in agents:
card = await _build_agent_card(agent)
cards.append(card)
# Filter by skill tag if specified
if skill:
cards = [
card
for card in cards
if any(skill.lower() in tag.lower() for s in card.skills for tag in s.tags)
]
return cards
@router.get("/agents/{agent_id}/card")
async def get_agent_card_by_id(
agent_id: str,
db: DbSession,
) -> AgentCard:
"""
Get Agent Card for a specific agent by ID or slug.
Alternative to the /.well-known/agent.json endpoint for programmatic access.
"""
# Try to parse as UUID first
try:
uuid = UUID(agent_id)
result = await db.execute(select(AgentTable).where(AgentTable.id == uuid))
except ValueError:
# Not a UUID, try slug lookup
result = await db.execute(
select(AgentTable).where(AgentTable.slug == agent_id)
)
agent = result.scalar_one_or_none()
if agent is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Agent not found: {agent_id}",
)
return await _build_agent_card(agent)
roboco/api/websocket_bridge.py
-160
View File
@@ -1,160 +0,0 @@
"""
WebSocket Event Bridge
Consumes events from Redis Streams and forwards them to WebSocket clients.
This enables real-time updates to connected clients while maintaining
durable message delivery through the stream.
"""
from uuid import UUID
import structlog
from roboco.api.websocket import broadcast_notification, manager
from roboco.events import Event, EventType, get_event_bus
logger = structlog.get_logger()
# Handler for notification events
async def _handle_notification_sent(event: Event) -> None:
"""Handle NOTIFICATION_SENT events and forward to WebSocket."""
data = event.data
notification_id_str = data.get("notification_id")
recipient_id_str = data.get("recipient_id")
notification_type = data.get("type", "unknown")
subject = data.get("subject", "")
priority = data.get("priority", "normal")
if not notification_id_str or not recipient_id_str:
logger.warning(
"Incomplete notification event",
event_id=str(event.id),
)
return
try:
notification_id = UUID(notification_id_str)
recipient_id = UUID(recipient_id_str)
except ValueError as e:
logger.error("Invalid UUID in notification event", error=str(e))
return
# Check if recipient has WebSocket connections
connections = manager.notification_connections.get(recipient_id, set())
if connections:
await broadcast_notification(
agent_ids=[recipient_id],
notification_id=notification_id,
notification_type=notification_type,
subject=subject,
priority=priority,
)
logger.debug(
"Notification forwarded to WebSocket",
notification_id=notification_id_str,
recipient=recipient_id_str,
connection_count=len(connections),
)
async def _handle_session_event(event: Event) -> None:
"""Handle session lifecycle events and forward to WebSocket."""
data = event.data
session_id_str = data.get("session_id")
if not session_id_str:
return
try:
session_id = UUID(session_id_str)
except ValueError:
return
connections = manager.session_connections.get(session_id, set())
if not connections:
return
# Forward event to session subscribers
event_payload = {
"type": f"session.{event.type.value.split('.')[-1]}",
"session_id": session_id_str,
"data": data,
}
await manager.broadcast_to_session(session_id, event_payload)
logger.debug(
"Session event forwarded to WebSocket",
event_type=event.type.value,
session_id=session_id_str,
)
async def _handle_agent_event(event: Event) -> None:
"""Handle agent lifecycle events and forward to WebSocket."""
data = event.data
agent_id_str = data.get("agent_id") or event.source_agent
if not agent_id_str:
return
try:
agent_id = UUID(agent_id_str)
except ValueError:
return
connections = manager.agent_connections.get(agent_id, set())
if not connections:
return
event_payload = {
"type": f"agent.{event.type.value.split('.')[-1]}",
"agent_id": agent_id_str,
"data": data,
}
await manager.broadcast_to_agent_watchers(agent_id, event_payload)
logger.debug(
"Agent event forwarded to WebSocket",
event_type=event.type.value,
agent_id=agent_id_str,
)
def register_websocket_bridge_handlers() -> None:
"""
Register event handlers that forward events to WebSocket clients.
Call this during application startup after the event bus is initialized.
"""
bus = get_event_bus()
# Notification events -> WebSocket
bus.subscribe(EventType.NOTIFICATION_SENT, _handle_notification_sent)
bus.subscribe(EventType.NOTIFICATION_ACKED, _handle_notification_sent)
# Session events -> WebSocket
bus.subscribe(EventType.SESSION_CREATED, _handle_session_event)
bus.subscribe(EventType.SESSION_CLOSED, _handle_session_event)
bus.subscribe(EventType.SESSION_TIMEOUT, _handle_session_event)
# Agent events -> WebSocket
bus.subscribe(EventType.AGENT_SPAWNED, _handle_agent_event)
bus.subscribe(EventType.AGENT_STOPPED, _handle_agent_event)
bus.subscribe(EventType.AGENT_WAITING, _handle_agent_event)
bus.subscribe(EventType.AGENT_RESUMED, _handle_agent_event)
bus.subscribe(EventType.AGENT_ERROR, _handle_agent_event)
logger.info("WebSocket bridge handlers registered")
async def start_websocket_bridge() -> None:
"""
Start the WebSocket bridge.
This registers handlers and ensures they're connected to the event stream.
Should be called during application startup.
"""
register_websocket_bridge_handlers()
logger.info("WebSocket bridge started")
roboco/bootstrap.py
+5 -12
View File
@@ -13,10 +13,9 @@ import uvicorn
from roboco.api.deps import set_orchestrator
from roboco.api.websocket import broadcast_agent_chunk
from roboco.api.websocket_bridge import start_websocket_bridge
from roboco.config import settings
from roboco.db import bootstrap_database
from roboco.events import init_event_bus, register_default_handlers, set_event_context
from roboco.events import EventBus, register_default_handlers, set_event_context
from roboco.runtime import AgentOrchestrator, set_reasoning_stream_callback
from roboco.services.notification import NotificationService
@@ -64,18 +63,12 @@ async def main(
logger.info("Orchestrator skipped, exiting")
return
# Initialize event bus (Redis Streams with consumer groups)
event_bus = await init_event_bus(
consumer_name=f"orchestrator-{settings.host}:{settings.port}",
recover_pending=True, # Recover unacknowledged messages from previous run
)
# Initialize event bus
event_bus = EventBus()
await event_bus.connect()
register_default_handlers(event_bus)
# Register WebSocket bridge handlers (forward stream events to WebSocket clients)
await start_websocket_bridge()
await event_bus.start_listening()
logger.info("Event bus initialized (Redis Streams)")
logger.info("Event bus initialized")
# Initialize orchestrator
orchestrator = AgentOrchestrator(
roboco/config.py
-33
View File
@@ -195,39 +195,6 @@ class Settings(BaseSettings):
session_max_content_length: int = Field(default=50000, ge=1)
message_max_length: int = Field(default=10000, ge=1)
# ==========================================================================
# Kubernetes
# ==========================================================================
k8s_enabled: bool = Field(
default=False, description="Enable K8s mode (use Jobs API instead of Docker)"
)
k8s_namespace: str = Field(default="roboco", description="K8s namespace for agents")
k8s_agent_image: str = Field(
default="ghcr.io/renzof/roboco-agent",
description="Base image for agent containers",
)
k8s_agent_image_tag: str = Field(
default="latest", description="Image tag for agent containers"
)
k8s_agent_memory_request: str = Field(
default="256Mi", description="Memory request for agent pods"
)
k8s_agent_memory_limit: str = Field(
default="1Gi", description="Memory limit for agent pods"
)
k8s_agent_cpu_request: str = Field(
default="100m", description="CPU request for agent pods"
)
k8s_agent_cpu_limit: str = Field(
default="500m", description="CPU limit for agent pods"
)
k8s_job_ttl: int = Field(
default=3600, ge=60, description="TTL in seconds for completed agent jobs"
)
k8s_job_backoff_limit: int = Field(
default=0, ge=0, description="Number of retries for failed jobs"
)
@lru_cache
def get_settings() -> Settings:
roboco/events/__init__.py
-6
View File
@@ -2,9 +2,6 @@
Event System for RoboCo
Handles workflow triggers and event-driven communication between components.
Uses Redis Streams for durable message delivery with consumer groups.
All events are persisted and delivered with at-least-once semantics.
"""
from roboco.events.bus import Event, EventBus, EventType, get_event_bus, init_event_bus
@@ -19,16 +16,13 @@ from roboco.events.handlers import (
register_default_handlers,
set_event_context,
)
from roboco.events.stream_bus import StreamEventBus, get_stream_event_bus
__all__ = [
"Event",
"EventBus",
"EventType",
"StreamEventBus",
"get_event_bus",
"get_event_context",
"get_stream_event_bus",
"handle_blocker_resolved",
"handle_handoff_created",
"handle_qa_result",
roboco/events/bus.py
+178 -38
View File
@@ -1,56 +1,196 @@
"""
Event Bus
Redis Streams-based event system for durable workflow triggers.
This module re-exports StreamEventBus as EventBus for backward compatibility.
All events are now persisted using Redis Streams with consumer groups for
guaranteed at-least-once delivery.
Redis-based pub/sub event system for workflow triggers.
"""
import asyncio
import contextlib
from collections.abc import Callable, Coroutine
from typing import Any
from roboco.events.stream_bus import (
StreamEventBus,
get_stream_event_bus,
init_stream_event_bus,
)
import redis.asyncio as redis
import structlog
from roboco.config import settings
from roboco.models.events import Event, EventType
# Type for event handlers (re-export for backward compatibility)
logger = structlog.get_logger()
# Type for event handlers
EventHandler = Callable[[Event], Coroutine[Any, Any, None]]
# Re-export StreamEventBus as EventBus for backward compatibility
EventBus = StreamEventBus
__all__ = [
"Event",
"EventBus",
"EventHandler",
"EventType",
"get_event_bus",
"init_event_bus",
class EventBus:
"""
Event bus for publishing and subscribing to events.
Uses Redis pub/sub for distributed event handling.
"""
CHANNEL_PREFIX = "roboco:events:"
def __init__(self, redis_url: str | None = None):
self.redis_url = redis_url or settings.redis_url
self._redis: redis.Redis | None = None
self._pubsub: redis.client.PubSub | None = None
self._handlers: dict[EventType, list[EventHandler]] = {}
self._running = False
self._listen_task: asyncio.Task | None = None
async def connect(self) -> None:
"""Connect to Redis."""
self._redis = redis.from_url(self.redis_url)
self._pubsub = self._redis.pubsub()
logger.info("EventBus connected to Redis")
async def disconnect(self) -> None:
"""Disconnect from Redis."""
self._running = False
if self._listen_task:
self._listen_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await self._listen_task
if self._pubsub:
await self._pubsub.close()
if self._redis:
await self._redis.close()
logger.info("EventBus disconnected")
def subscribe(self, event_type: EventType, handler: EventHandler) -> None:
"""Subscribe a handler to an event type."""
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
logger.debug("Handler subscribed", event_type=event_type.value)
def unsubscribe(self, event_type: EventType, handler: EventHandler) -> None:
"""Unsubscribe a handler from an event type."""
if event_type in self._handlers:
self._handlers[event_type] = [
h for h in self._handlers[event_type] if h != handler
]
async def publish(self, event: Event) -> None:
"""Publish an event."""
if not self._redis:
raise RuntimeError("EventBus not connected")
def get_event_bus() -> StreamEventBus:
"""Get or create the global event bus instance."""
return get_stream_event_bus()
channel = f"{self.CHANNEL_PREFIX}{event.type.value}"
await self._redis.publish(channel, event.to_json())
async def init_event_bus(
consumer_name: str | None = None,
recover_pending: bool = True,
) -> StreamEventBus:
"""
Initialize and start the event bus.
Args:
consumer_name: Unique name for this consumer instance
recover_pending: Whether to recover unacknowledged messages on startup
"""
return await init_stream_event_bus(
consumer_name=consumer_name,
recover_pending=recover_pending,
logger.info(
"Event published",
event_type=event.type.value,
event_id=str(event.id),
source=event.source_agent,
)
async def publish_task_event(
self,
event_type: EventType,
task_id: str,
agent_id: str | None = None,
**extra_data: Any,
) -> None:
"""Convenience method to publish task-related events."""
event = Event(
type=event_type,
data={"task_id": task_id, **extra_data},
source_agent=agent_id,
)
await self.publish(event)
async def start_listening(self) -> None:
"""Start listening for events."""
if not self._pubsub:
raise RuntimeError("EventBus not connected")
# Subscribe to all event channels we have handlers for
channels = [f"{self.CHANNEL_PREFIX}{et.value}" for et in self._handlers]
if not channels:
logger.warning("No event handlers registered, nothing to subscribe to")
return
await self._pubsub.subscribe(*channels)
self._running = True
self._listen_task = asyncio.create_task(self._listen_loop())
logger.info("EventBus listening", channels=len(channels))
async def _listen_loop(self) -> None:
"""Main event listening loop."""
while self._running:
try:
if self._pubsub is None:
break
message = await self._pubsub.get_message(
ignore_subscribe_messages=True,
timeout=1.0,
)
if message and message["type"] == "message":
await self._handle_message(message)
except asyncio.CancelledError:
break
except Exception as e:
logger.error("Error in event loop", error=str(e))
await asyncio.sleep(1)
async def _handle_message(self, message: dict) -> None:
"""Handle an incoming message."""
try:
event = Event.from_json(message["data"])
handlers = self._handlers.get(event.type, [])
if not handlers:
return
logger.debug(
"Handling event",
event_type=event.type.value,
handler_count=len(handlers),
)
# Run all handlers concurrently
tasks = [handler(event) for handler in handlers]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Log any errors
for i, result in enumerate(results):
if isinstance(result, Exception):
logger.error(
"Event handler error",
event_type=event.type.value,
handler=handlers[i].__name__,
error=str(result),
)
except Exception as e:
logger.error("Failed to handle message", error=str(e))
class _EventBusHolder:
"""Holder for singleton EventBus instance."""
instance: EventBus | None = None
def get_event_bus() -> EventBus:
"""Get or create the global event bus instance."""
if _EventBusHolder.instance is None:
_EventBusHolder.instance = EventBus()
return _EventBusHolder.instance
async def init_event_bus() -> EventBus:
"""Initialize and start the event bus."""
bus = get_event_bus()
await bus.connect()
return bus
roboco/events/stream_bus.py
-401
View File
@@ -1,401 +0,0 @@
"""
Stream Event Bus
Redis Streams-based event system with durable message delivery.
Replaces the pub/sub-based EventBus with persistence and consumer groups.
"""
import asyncio
import contextlib
from collections.abc import Callable, Coroutine
from typing import Any
import redis.asyncio as redis
import structlog
from redis.exceptions import ResponseError
from roboco.config import settings
from roboco.models.events import Event, EventType
logger = structlog.get_logger()
# Type for event handlers
EventHandler = Callable[[Event], Coroutine[Any, Any, None]]
class StreamEventBus:
"""
Event bus using Redis Streams for durable message delivery.
Features:
- Message persistence (survives Redis restart with AOF)
- Consumer groups for at-least-once delivery
- Message acknowledgment after successful processing
- Automatic stream trimming (configurable retention)
"""
STREAM_PREFIX = "roboco:stream:"
DEFAULT_GROUP = "roboco-handlers"
MAX_STREAM_LENGTH = 10000 # Trim streams to this length
def __init__(
self,
redis_url: str | None = None,
consumer_name: str | None = None,
group_name: str | None = None,
):
self.redis_url = redis_url or settings.redis_url
self.consumer_name = consumer_name or f"consumer-{id(self)}"
self.group_name = group_name or self.DEFAULT_GROUP
self._redis: redis.Redis | None = None
self._handlers: dict[EventType, list[EventHandler]] = {}
self._running = False
self._listen_task: asyncio.Task | None = None
async def connect(self) -> None:
"""Connect to Redis."""
self._redis = redis.from_url(self.redis_url)
logger.info("StreamEventBus connected to Redis")
def is_connected(self) -> bool:
"""Check if the event bus is connected to Redis."""
return self._redis is not None
async def disconnect(self) -> None:
"""Disconnect from Redis."""
self._running = False
if self._listen_task:
self._listen_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await self._listen_task
if self._redis:
await self._redis.close()
logger.info("StreamEventBus disconnected")
def _get_stream_name(self, event_type: EventType) -> str:
"""Get stream name for event type (grouped by prefix)."""
# Group by event category: task.*, agent.*, notification.*, etc.
category = event_type.value.split(".")[0]
return f"{self.STREAM_PREFIX}{category}"
def _get_all_stream_names(self) -> list[str]:
"""Get all stream names for registered handlers."""
categories = set()
for event_type in self._handlers:
category = event_type.value.split(".")[0]
categories.add(category)
return [f"{self.STREAM_PREFIX}{cat}" for cat in categories]
async def _ensure_consumer_group(self, stream: str) -> None:
"""Ensure consumer group exists for stream."""
if not self._redis:
return
try:
await self._redis.xgroup_create(
stream,
self.group_name,
id="0",
mkstream=True,
)
logger.debug("Created consumer group", stream=stream, group=self.group_name)
except ResponseError as e:
if "BUSYGROUP" not in str(e):
raise
# Group already exists, that's fine
def subscribe(self, event_type: EventType, handler: EventHandler) -> None:
"""Subscribe a handler to an event type."""
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
logger.debug("Handler subscribed", event_type=event_type.value)
def unsubscribe(self, event_type: EventType, handler: EventHandler) -> None:
"""Unsubscribe a handler from an event type."""
if event_type in self._handlers:
self._handlers[event_type] = [
h for h in self._handlers[event_type] if h != handler
]
async def publish(self, event: Event) -> str:
"""
Publish an event to the stream.
Returns the message ID assigned by Redis.
"""
if not self._redis:
raise RuntimeError("StreamEventBus not connected")
stream = self._get_stream_name(event.type)
# Add to stream with automatic ID (*) and trim to max length
raw_message_id = await self._redis.xadd(
stream,
{
"type": event.type.value,
"data": event.to_json(),
},
maxlen=self.MAX_STREAM_LENGTH,
approximate=True,
)
# Convert bytes to str if needed
message_id = (
raw_message_id.decode()
if isinstance(raw_message_id, bytes)
else str(raw_message_id)
)
logger.info(
"Event published to stream",
event_type=event.type.value,
event_id=str(event.id),
stream=stream,
message_id=message_id,
source=event.source_agent,
)
return message_id
async def publish_task_event(
self,
event_type: EventType,
task_id: str,
agent_id: str | None = None,
**extra_data: Any,
) -> str:
"""Convenience method to publish task-related events."""
event = Event(
type=event_type,
data={"task_id": task_id, **extra_data},
source_agent=agent_id,
)
return await self.publish(event)
async def start_listening(self) -> None:
"""Start listening for events."""
if not self._redis:
raise RuntimeError("StreamEventBus not connected")
streams = self._get_all_stream_names()
if not streams:
logger.warning("No event handlers registered, nothing to subscribe to")
return
# Ensure consumer groups exist for all streams
for stream in streams:
await self._ensure_consumer_group(stream)
self._running = True
self._listen_task = asyncio.create_task(self._listen_loop())
logger.info("StreamEventBus listening", streams=streams)
async def _listen_loop(self) -> None:
"""Main event listening loop using XREADGROUP."""
if not self._redis:
return
streams = self._get_all_stream_names()
# Build stream dict: {stream_name: ">"} (> = only new messages)
stream_dict = dict.fromkeys(streams, ">")
while self._running:
try:
# Block for 5 seconds waiting for new messages
results = await self._redis.xreadgroup(
self.group_name,
self.consumer_name,
stream_dict,
count=10,
block=5000,
)
if not results:
continue
for stream_name, messages in results:
for message_id, data in messages:
await self._handle_message(stream_name, message_id, data)
except asyncio.CancelledError:
break
except Exception as e:
logger.error("Error in stream event loop", error=str(e))
await asyncio.sleep(1)
async def _handle_message(
self,
stream: str,
message_id: str,
data: dict,
) -> None:
"""Handle an incoming message and ACK on success."""
if not self._redis:
return
try:
event_data = data.get(b"data") or data.get("data")
if isinstance(event_data, bytes):
event_data = event_data.decode()
if not event_data or not isinstance(event_data, str):
logger.error("Invalid event data", message_id=message_id)
await self._redis.xack(stream, self.group_name, message_id)
return
event = Event.from_json(event_data)
handlers = self._handlers.get(event.type, [])
if not handlers:
# No handlers but still ACK to prevent redelivery
await self._redis.xack(stream, self.group_name, message_id)
return
logger.debug(
"Handling event from stream",
event_type=event.type.value,
message_id=message_id,
handler_count=len(handlers),
)
# Run all handlers concurrently
tasks = [handler(event) for handler in handlers]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Log any errors
all_succeeded = True
for i, result in enumerate(results):
if isinstance(result, Exception):
all_succeeded = False
logger.error(
"Event handler error",
event_type=event.type.value,
handler=handlers[i].__name__,
error=str(result),
)
# ACK the message if all handlers succeeded
# If any failed, message stays pending and can be reclaimed later
if all_succeeded:
await self._redis.xack(stream, self.group_name, message_id)
logger.debug("Message acknowledged", message_id=message_id)
else:
logger.warning(
"Message not acknowledged due to handler errors",
message_id=message_id,
)
except Exception as e:
logger.error(
"Failed to handle stream message",
error=str(e),
message_id=message_id,
)
async def recover_pending(self, idle_time_ms: int = 60000) -> int:
"""
Recover pending messages that weren't acknowledged.
Useful for startup to process messages from crashed consumers.
Args:
idle_time_ms: Only recover messages idle for this long (default 1 minute)
Returns:
Number of messages recovered
"""
if not self._redis:
return 0
recovered = 0
streams = self._get_all_stream_names()
for stream in streams:
try:
# Claim pending messages from any consumer
pending = await self._redis.xpending(stream, self.group_name)
if not pending or pending["pending"] == 0:
continue
# Get pending message details
pending_details = await self._redis.xpending_range(
stream,
self.group_name,
min="-",
max="+",
count=100,
)
for msg in pending_details:
msg_id = msg["message_id"]
idle = msg["time_since_delivered"]
if idle >= idle_time_ms:
# Claim message for this consumer
claimed = await self._redis.xclaim(
stream,
self.group_name,
self.consumer_name,
min_idle_time=idle_time_ms,
message_ids=[msg_id],
)
if claimed:
recovered += 1
# Process the claimed message
for claim_id, data in claimed:
await self._handle_message(stream, claim_id, data)
except Exception as e:
logger.error(
"Error recovering pending messages",
stream=stream,
error=str(e),
)
if recovered:
logger.info("Recovered pending messages", count=recovered)
return recovered
# =============================================================================
# SINGLETON ACCESS
# =============================================================================
class _StreamEventBusHolder:
"""Holder for singleton StreamEventBus instance."""
instance: StreamEventBus | None = None
def get_stream_event_bus() -> StreamEventBus:
"""Get or create the global stream event bus instance."""
if _StreamEventBusHolder.instance is None:
_StreamEventBusHolder.instance = StreamEventBus()
return _StreamEventBusHolder.instance
async def init_stream_event_bus(
consumer_name: str | None = None,
recover_pending: bool = True,
) -> StreamEventBus:
"""
Initialize and start the stream event bus.
Args:
consumer_name: Unique name for this consumer instance
recover_pending: Whether to recover unacknowledged messages on startup
"""
bus = get_stream_event_bus()
if consumer_name:
bus.consumer_name = consumer_name
await bus.connect()
if recover_pending:
await bus.recover_pending()
return bus
roboco/models/a2a.py
-447
View File
@@ -1,447 +0,0 @@
"""
A2A (Agent-to-Agent) Protocol Models
Implements Google's A2A protocol for agent interoperability.
See: https://a2a-protocol.org/latest/specification/
This module defines the data structures for:
- AgentCard: Agent metadata and capability discovery
- Task: Work unit lifecycle management
- Message: Communication between agents
- Skill: Capability units an agent can perform
"""
from datetime import UTC, datetime
from enum import Enum
from typing import Annotated, Any, Literal
from uuid import uuid4
from pydantic import ConfigDict, Field
from roboco.models.base import RobocoBase
# =============================================================================
# ENUMS
# =============================================================================
class A2ATaskState(str, Enum):
"""
A2A Task lifecycle states.
Per A2A specification section 4.2.1.
"""
SUBMITTED = "submitted" # Task acknowledged and created
WORKING = "working" # Actively processing
COMPLETED = "completed" # Finished successfully (terminal)
FAILED = "failed" # Finished with error (terminal)
CANCELLED = "cancelled" # Stopped before completion (terminal)
INPUT_REQUIRED = "input_required" # Awaiting additional information
REJECTED = "rejected" # Agent declined the task (terminal)
AUTH_REQUIRED = "auth_required" # Needs client authentication
class A2APartType(str, Enum):
"""Types of content parts in a message."""
TEXT = "text"
FILE = "file"
DATA = "data"
ARTIFACT = "artifact"
# =============================================================================
# AGENT CARD MODELS
# =============================================================================
class AgentProvider(RobocoBase):
"""Provider information for an agent."""
organization: str = Field(..., description="Organization name")
url: str | None = Field(default=None, description="Organization URL")
class AgentCapabilities(RobocoBase):
"""Capabilities supported by an agent."""
streaming: bool = Field(default=False, description="Supports SSE streaming")
push_notifications: bool = Field(
default=False, description="Supports webhook push notifications"
)
state_transition_history: bool = Field(
default=False, description="Returns task state history"
)
class SecurityScheme(RobocoBase):
"""Security scheme for authentication."""
type: str = Field(..., description="Scheme type (apiKey, http, oauth2)")
name: str | None = Field(default=None, description="Name of the header/param")
scheme: str | None = Field(default=None, description="HTTP auth scheme (bearer)")
bearer_format: str | None = Field(
default=None, alias="bearerFormat", description="Format hint for tokens"
)
in_location: str | None = Field(
default=None, alias="in", description="Where to send (header, query, cookie)"
)
class AgentSkill(RobocoBase):
"""A capability unit an agent can perform."""
id: str = Field(..., description="Unique skill identifier")
name: str = Field(..., description="Human-readable skill name")
description: str = Field(..., description="What this skill does")
tags: list[str] = Field(default_factory=list, description="Capability categories")
examples: list[str] = Field(
default_factory=list, description="Example invocations"
)
input_modes: list[str] = Field(
default_factory=lambda: ["text/plain"],
alias="inputModes",
description="Supported input MIME types",
)
output_modes: list[str] = Field(
default_factory=lambda: ["text/plain"],
alias="outputModes",
description="Supported output MIME types",
)
class AgentCard(RobocoBase):
"""
A2A Agent Card - The agent's public identity and capabilities.
Published at /.well-known/agent.json per A2A specification.
Acts as the agent's "business card" for discovery.
"""
# Required fields
id: str = Field(..., description="Unique agent identifier")
name: str = Field(..., description="Human-readable agent name")
provider: AgentProvider = Field(..., description="Provider information")
protocol_version: str = Field(
default="1.0",
alias="protocolVersion",
description="Supported A2A protocol version",
)
service_endpoint: str = Field(
..., alias="serviceEndpoint", description="Base URL for A2A operations"
)
capabilities: AgentCapabilities = Field(
default_factory=AgentCapabilities, description="Supported features"
)
security_schemes: dict[str, SecurityScheme] = Field(
default_factory=dict,
alias="securitySchemes",
description="Available auth methods",
)
security: list[dict[str, list[str]]] = Field(
default_factory=list, description="Required security scheme(s)"
)
# Optional fields
description: str | None = Field(
default=None, description="Agent purpose and capabilities"
)
skills: list[AgentSkill] = Field(
default_factory=list, description="Available operations"
)
default_input_modes: list[str] = Field(
default_factory=lambda: ["text/plain", "application/json"],
alias="defaultInputModes",
description="Default accepted input MIME types",
)
default_output_modes: list[str] = Field(
default_factory=lambda: ["text/plain", "application/json"],
alias="defaultOutputModes",
description="Default output MIME types",
)
documentation_url: str | None = Field(
default=None, alias="documentationUrl", description="Agent documentation URL"
)
version: str | None = Field(default=None, description="Agent version")
supports_extended_agent_card: bool = Field(
default=False,
alias="supportsExtendedAgentCard",
description="Whether authenticated card available",
)
metadata: dict[str, Any] = Field(
default_factory=dict, description="Custom metadata"
)
model_config = ConfigDict(
populate_by_name=True,
extra="allow", # A2A cards may have extensions
)
# =============================================================================
# MESSAGE MODELS
# =============================================================================
class TextPart(RobocoBase):
"""Plain text content part."""
type: Literal["text"] = "text"
text: str = Field(..., description="Text content")
class FilePart(RobocoBase):
"""File reference content part."""
type: Literal["file"] = "file"
file: dict[str, Any] = Field(
..., description="File data (uri, mimeType, name, data)"
)
class DataPart(RobocoBase):
"""Structured JSON data content part."""
type: Literal["data"] = "data"
data: dict[str, Any] = Field(..., description="Structured data")
class ArtifactPart(RobocoBase):
"""Reference to a generated artifact."""
type: Literal["artifact"] = "artifact"
artifact: dict[str, Any] = Field(..., description="Artifact reference")
# Union type for message parts
Part = Annotated[
TextPart | FilePart | DataPart | ArtifactPart, Field(discriminator="type")
]
class A2AMessage(RobocoBase):
"""
A2A Message - A communication turn between agents.
Contains one or more parts with content.
"""
role: Literal["user", "agent"] = Field(
..., description="Message sender role"
)
parts: list[Part] = Field(..., description="Content parts")
context_id: str | None = Field(
default=None, alias="contextId", description="Conversation grouping"
)
task_id: str | None = Field(
default=None, alias="taskId", description="Associated task"
)
message_id: str = Field(
default_factory=lambda: str(uuid4()),
alias="messageId",
description="Unique message ID",
)
model_config = ConfigDict(populate_by_name=True)
# =============================================================================
# TASK MODELS
# =============================================================================
class A2ATaskStatus(RobocoBase):
"""Status container for an A2A task."""
state: A2ATaskState = Field(..., description="Current lifecycle state")
message: A2AMessage | None = Field(
default=None, description="Associated status message"
)
timestamp: datetime = Field(
default_factory=lambda: datetime.now(UTC),
description="When status was recorded",
)
class A2AArtifact(RobocoBase):
"""An output artifact produced by a task."""
id: str = Field(default_factory=lambda: str(uuid4()), description="Artifact ID")
name: str = Field(..., description="Artifact name")
parts: list[Part] = Field(..., description="Artifact content parts")
metadata: dict[str, Any] = Field(
default_factory=dict, description="Custom metadata"
)
class A2ATask(RobocoBase):
"""
A2A Task - A unit of work with lifecycle management.
Maps to RoboCo's internal TaskTable but follows A2A semantics.
"""
id: str = Field(
default_factory=lambda: str(uuid4()), description="Server-generated task ID"
)
context_id: str = Field(
..., alias="contextId", description="Groups related interactions"
)
status: A2ATaskStatus = Field(..., description="Current task status")
artifacts: list[A2AArtifact] = Field(
default_factory=list, description="Output artifacts"
)
history: list[A2AMessage] = Field(
default_factory=list, description="Interaction history"
)
metadata: dict[str, Any] = Field(
default_factory=dict, description="Custom metadata"
)
model_config = ConfigDict(populate_by_name=True)
# =============================================================================
# JSON-RPC REQUEST/RESPONSE MODELS
# =============================================================================
class SendMessageConfiguration(RobocoBase):
"""Configuration for SendMessage request."""
accepted_output_modes: list[str] = Field(
default_factory=lambda: ["text/plain", "application/json"],
alias="acceptedOutputModes",
description="Client-accepted output MIME types",
)
history_length: int | None = Field(
default=None,
alias="historyLength",
description="Number of history turns to include in response",
)
blocking: bool = Field(
default=False, description="Wait for task completion before responding"
)
push_notification_config: dict[str, Any] | None = Field(
default=None,
alias="pushNotificationConfig",
description="Webhook config for async updates",
)
model_config = ConfigDict(populate_by_name=True)
class SendMessageRequest(RobocoBase):
"""Request payload for SendMessage JSON-RPC method."""
message: A2AMessage = Field(..., description="Message to send")
configuration: SendMessageConfiguration | None = Field(
default=None, description="Request configuration"
)
metadata: dict[str, Any] = Field(
default_factory=dict, description="Custom request metadata"
)
class SendMessageResponse(RobocoBase):
"""Response payload for SendMessage JSON-RPC method."""
task: A2ATask = Field(..., description="Created or updated task")
class GetTaskRequest(RobocoBase):
"""Request payload for GetTask JSON-RPC method."""
name: str = Field(..., description="Task resource name (tasks/{id})")
history_length: int | None = Field(
default=None,
alias="historyLength",
description="Number of history turns to include",
)
model_config = ConfigDict(populate_by_name=True)
class ListTasksRequest(RobocoBase):
"""Request payload for ListTasks JSON-RPC method."""
page_size: int = Field(
default=20, alias="pageSize", ge=1, le=100, description="Results per page"
)
page_token: str | None = Field(
default=None, alias="pageToken", description="Pagination token"
)
filter: str | None = Field(default=None, description="Filter expression")
order_by: str | None = Field(
default=None, alias="orderBy", description="Sort order"
)
model_config = ConfigDict(populate_by_name=True)
class ListTasksResponse(RobocoBase):
"""Response payload for ListTasks JSON-RPC method."""
tasks: list[A2ATask] = Field(..., description="Task list")
next_page_token: str | None = Field(
default=None, alias="nextPageToken", description="Token for next page"
)
model_config = ConfigDict(populate_by_name=True)
class CancelTaskRequest(RobocoBase):
"""Request payload for CancelTask JSON-RPC method."""
name: str = Field(..., description="Task resource name (tasks/{id})")
reason: str | None = Field(default=None, description="Cancellation reason")
# =============================================================================
# STATE MAPPING UTILITIES
# =============================================================================
def task_status_to_a2a_state(roboco_status: str) -> A2ATaskState:
"""
Map RoboCo TaskStatus to A2A TaskState.
This enables interoperability between RoboCo's internal
task lifecycle and the A2A protocol.
"""
mapping = {
"backlog": A2ATaskState.SUBMITTED,
"pending": A2ATaskState.SUBMITTED,
"claimed": A2ATaskState.WORKING,
"in_progress": A2ATaskState.WORKING,
"blocked": A2ATaskState.INPUT_REQUIRED,
"paused": A2ATaskState.INPUT_REQUIRED,
"verifying": A2ATaskState.WORKING,
"needs_revision": A2ATaskState.INPUT_REQUIRED,
"awaiting_qa": A2ATaskState.WORKING,
"awaiting_documentation": A2ATaskState.WORKING,
"awaiting_pm_review": A2ATaskState.WORKING,
"completed": A2ATaskState.COMPLETED,
"cancelled": A2ATaskState.CANCELLED,
}
return mapping.get(roboco_status, A2ATaskState.WORKING)
def a2a_state_to_task_status(a2a_state: A2ATaskState) -> str:
"""
Map A2A TaskState back to RoboCo TaskStatus.
Used when creating tasks via A2A protocol.
"""
mapping = {
A2ATaskState.SUBMITTED: "pending",
A2ATaskState.WORKING: "in_progress",
A2ATaskState.COMPLETED: "completed",
A2ATaskState.FAILED: "cancelled", # RoboCo uses cancelled for failures
A2ATaskState.CANCELLED: "cancelled",
A2ATaskState.INPUT_REQUIRED: "blocked",
A2ATaskState.REJECTED: "cancelled",
A2ATaskState.AUTH_REQUIRED: "blocked",
}
return mapping.get(a2a_state, "pending")
roboco/runtime/orchestrator.py
+12 -401
View File
@@ -10,10 +10,6 @@ The orchestrator is the BRAIN of the system:
- Agents receive their assignment at spawn time
- Agents scan for more work after completing a task
- Agents only call roboco_agent_idle() when truly no work remains
Supports two backends:
- Docker: Uses Docker CLI to spawn containers (default)
- Kubernetes: Uses K8s Jobs API to spawn agent pods (when k8s_enabled=True)
"""
import asyncio
@@ -23,7 +19,7 @@ import os
import tempfile
from datetime import UTC, datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any, ClassVar
from typing import Any, ClassVar
import httpx
import structlog
@@ -43,10 +39,6 @@ from roboco.models.runtime import (
)
from roboco.seeds.initial_data import AGENT_UUIDS
# K8s imports (only when K8s mode is enabled)
if TYPE_CHECKING:
from kubernetes import client as k8s_client
logger = structlog.get_logger()
# Reverse mapping: UUID -> slug
@@ -114,15 +106,11 @@ class AgentOrchestrator:
Manages Claude Code containers for all agents.
Responsibilities:
- Spawn agents as Docker containers or K8s Jobs
- Monitor health via docker inspect or K8s API
- Spawn agents as Docker containers
- Monitor health via docker inspect
- Handle waiting states and respawning
- Provide status API
- Cost-efficient on-demand spawning
Backends:
- Docker (default): Uses Docker CLI to spawn containers
- Kubernetes: Uses K8s Jobs API when settings.k8s_enabled=True
"""
def __init__(
@@ -144,51 +132,15 @@ class AgentOrchestrator:
self._running = False
self._lock = asyncio.Lock()
# K8s clients (initialized lazily when k8s_enabled)
self._k8s_batch_v1: k8s_client.BatchV1Api | None = None
self._k8s_core_v1: k8s_client.CoreV1Api | None = None
self._k8s_initialized = False
# =========================================================================
# LIFECYCLE
# =========================================================================
def _init_k8s(self) -> None:
"""Initialize Kubernetes clients."""
if self._k8s_initialized:
return
from kubernetes import client, config
# Load in-cluster config (when running in K8s)
# or kubeconfig for local dev
try:
config.load_incluster_config()
logger.info("K8s: loaded in-cluster config")
except config.ConfigException:
try:
config.load_kube_config()
logger.info("K8s: loaded kubeconfig")
except config.ConfigException:
logger.warning("K8s: no config found, K8s mode disabled")
return
self._k8s_batch_v1 = client.BatchV1Api()
self._k8s_core_v1 = client.CoreV1Api()
self._k8s_initialized = True
logger.info("K8s clients initialized", namespace=settings.k8s_namespace)
async def start(self) -> None:
"""Start the orchestrator."""
self._running = True
# Initialize K8s if enabled
if settings.k8s_enabled:
self._init_k8s()
if not self._k8s_initialized:
logger.warning("K8s mode requested but initialization failed")
else:
# Docker mode: Ensure agent image is built
# Ensure agent image is built
await self._ensure_agent_image()
# Ensure agent Claude settings have MCP tools allowed
@@ -202,7 +154,6 @@ class AgentOrchestrator:
"Orchestrator started",
dispatcher_interval=self.dispatcher_interval,
internal_api_url=self._api_url,
k8s_enabled=settings.k8s_enabled,
)
async def stop(self) -> None:
@@ -384,8 +335,6 @@ class AgentOrchestrator:
"""
Spawn a Claude Code container for an agent.
Uses Docker containers by default, or K8s Jobs when k8s_enabled=True.
Args:
agent_id: Agent identifier (e.g., "be-dev-1")
initial_prompt: Optional initial prompt
@@ -413,8 +362,7 @@ class AgentOrchestrator:
# Ensure agent Claude settings have MCP tools allowed
self._ensure_agent_claude_settings()
# Docker mode: Ensure agent-specific Docker image is built
if not settings.k8s_enabled:
# Ensure agent-specific Docker image is built
await self._ensure_agent_image(agent_id)
# Generate MCP config
@@ -443,34 +391,20 @@ class AgentOrchestrator:
self._instances[agent_id] = instance
# Spawn the container/job
# Spawn the container
try:
if settings.k8s_enabled and self._k8s_initialized:
# K8s mode: Create ConfigMaps and spawn Job
system_prompt = blueprint_path.read_text()
mcp_config_data = json.loads(mcp_config_path.read_text())
await self._create_k8s_agent_configmaps(
agent_id, system_prompt, mcp_config_data
)
job_name = await self._spawn_k8s_job(config, initial_prompt, task_id)
instance.container_id = job_name # Store job name as container_id
else:
# Docker mode: Spawn container
container_id = await self._spawn_container(config, initial_prompt)
instance.container_id = container_id
instance.state = AgentState.ACTIVE
instance.started_at = datetime.now(UTC)
instance.last_activity = datetime.now(UTC)
cid = instance.container_id[:12] if instance.container_id else None
logger.info(
"Agent spawned",
agent_id=agent_id,
container_id=cid,
container_id=container_id[:12],
model=model,
task_id=task_id,
k8s_enabled=settings.k8s_enabled,
)
return instance
@@ -613,315 +547,6 @@ class AgentOrchestrator:
)
await proc.wait()
# =========================================================================
# K8S JOB MANAGEMENT
# =========================================================================
async def _spawn_k8s_job(
self,
config: AgentConfig,
initial_prompt: str | None = None,
task_id: str | None = None,
) -> str:
"""Spawn an agent as a K8s Job."""
from kubernetes import client
if not self._k8s_batch_v1:
raise RuntimeError("K8s client not initialized")
job_name = f"agent-{config.agent_id}"
if task_id:
job_name = f"{job_name}-{task_id[:8]}"
# Delete existing job if any
await self._delete_k8s_job(job_name)
# Build environment variables
env_vars = [
client.V1EnvVar(name="ROBOCO_AGENT_ID", value=config.agent_id),
client.V1EnvVar(name="ROBOCO_API_URL", value="http://roboco-api:8000"),
client.V1EnvVar(
name="ROBOCO_ANTHROPIC_API_KEY",
value_from=client.V1EnvVarSource(
secret_key_ref=client.V1SecretKeySelector(
name="roboco-secrets",
key="anthropic-api-key",
),
),
),
]
if task_id:
env_vars.append(client.V1EnvVar(name="ROBOCO_TASK_ID", value=task_id))
# Build command with initial prompt
cmd_prompt = initial_prompt or (
"You may have been spawned without a specific task assignment. "
"Follow your standard workflow:\n\n"
"1. Call `roboco_task_scan()` to find available work for your role\n"
"2. If tasks are found, claim one with `roboco_task_claim(task_id)` "
"and begin the full task lifecycle "
"(UNDERSTAND -> PLAN -> EXECUTE -> VERIFY -> HANDOFF)\n"
"3. If no tasks are available, call `roboco_agent_idle()` "
"to shutdown gracefully\n\n"
"Start now by scanning for work."
)
# Agent image based on role
agent_image = f"{settings.k8s_agent_image}:{settings.k8s_agent_image_tag}"
# Build Job spec
job = client.V1Job(
api_version="batch/v1",
kind="Job",
metadata=client.V1ObjectMeta(
name=job_name,
namespace=settings.k8s_namespace,
labels={
"app": "roboco-agent",
"agent-id": config.agent_id,
"task-id": task_id[:8] if task_id else "none",
},
),
spec=client.V1JobSpec(
ttl_seconds_after_finished=settings.k8s_job_ttl,
backoff_limit=settings.k8s_job_backoff_limit,
template=client.V1PodTemplateSpec(
metadata=client.V1ObjectMeta(
labels={
"app": "roboco-agent",
"agent-id": config.agent_id,
},
),
spec=client.V1PodSpec(
restart_policy="Never",
service_account_name="roboco-agent",
containers=[
client.V1Container(
name="agent",
image=agent_image,
args=[
"--model",
MODEL_MAP.get(config.model, config.model),
"--system-prompt-file",
"/app/system-prompt.md",
"--mcp-config",
"/app/mcp-config.json",
"--output-format",
"stream-json",
"--verbose",
"-p",
cmd_prompt,
],
env=env_vars,
resources=client.V1ResourceRequirements(
requests={
"memory": settings.k8s_agent_memory_request,
"cpu": settings.k8s_agent_cpu_request,
},
limits={
"memory": settings.k8s_agent_memory_limit,
"cpu": settings.k8s_agent_cpu_limit,
},
),
volume_mounts=[
client.V1VolumeMount(
name="system-prompt",
mount_path="/app/system-prompt.md",
sub_path="system-prompt.md",
read_only=True,
),
client.V1VolumeMount(
name="mcp-config",
mount_path="/app/mcp-config.json",
sub_path="mcp-config.json",
read_only=True,
),
client.V1VolumeMount(
name="claude-auth",
mount_path="/home/agent/.claude",
),
],
),
],
volumes=[
client.V1Volume(
name="system-prompt",
config_map=client.V1ConfigMapVolumeSource(
name=f"agent-prompt-{config.agent_id}",
),
),
client.V1Volume(
name="mcp-config",
config_map=client.V1ConfigMapVolumeSource(
name=f"agent-mcp-{config.agent_id}",
),
),
client.V1Volume(
name="claude-auth",
secret=client.V1SecretVolumeSource(
secret_name="claude-auth",
optional=True,
),
),
],
),
),
),
)
# Create the Job
self._k8s_batch_v1.create_namespaced_job(
namespace=settings.k8s_namespace,
body=job,
)
logger.info(
"K8s: spawned agent job",
job_name=job_name,
agent_id=config.agent_id,
task_id=task_id,
)
return job_name
async def _delete_k8s_job(self, job_name: str) -> None:
"""Delete a K8s Job if it exists."""
from kubernetes.client.rest import ApiException
if not self._k8s_batch_v1:
return
try:
self._k8s_batch_v1.delete_namespaced_job(
name=job_name,
namespace=settings.k8s_namespace,
propagation_policy="Background",
)
logger.debug("K8s: deleted job", job_name=job_name)
except ApiException as e:
# 404 = not found, which is expected when job doesn't exist
if e.status != 404: # noqa: PLR2004
logger.warning("K8s: failed to delete job", job_name=job_name, error=e)
async def _stop_k8s_agent(self, agent_id: str) -> None:
"""Stop an agent by deleting its K8s Job."""
from kubernetes.client.rest import ApiException
if not self._k8s_batch_v1:
return
# Find jobs for this agent
try:
jobs = self._k8s_batch_v1.list_namespaced_job(
namespace=settings.k8s_namespace,
label_selector=f"agent-id={agent_id}",
)
for job in jobs.items:
await self._delete_k8s_job(job.metadata.name)
except ApiException as e:
logger.warning(
"K8s: failed to list jobs for agent", agent_id=agent_id, error=e
)
async def _get_k8s_agent_status( # noqa: PLR0911
self, agent_id: str
) -> AgentState:
"""Get agent status from K8s Job status."""
from kubernetes.client.rest import ApiException
if not self._k8s_batch_v1:
return AgentState.OFFLINE
try:
jobs = self._k8s_batch_v1.list_namespaced_job(
namespace=settings.k8s_namespace,
label_selector=f"agent-id={agent_id}",
)
if not jobs.items:
return AgentState.OFFLINE
# Get most recent job
job = sorted(
jobs.items,
key=lambda j: j.metadata.creation_timestamp or datetime.min,
reverse=True,
)[0]
if job.status.succeeded and job.status.succeeded > 0:
return AgentState.OFFLINE # Completed successfully
if job.status.failed and job.status.failed > 0:
return AgentState.OFFLINE # Failed
if job.status.active and job.status.active > 0:
return AgentState.ACTIVE
return AgentState.STARTING
except ApiException as e:
logger.warning(
"K8s: failed to get job status", agent_id=agent_id, error=e
)
return AgentState.OFFLINE
async def _create_k8s_agent_configmaps(
self,
agent_id: str,
system_prompt: str,
mcp_config: dict[str, Any],
) -> None:
"""Create ConfigMaps for agent prompt and MCP config."""
from kubernetes import client
from kubernetes.client.rest import ApiException
if not self._k8s_core_v1:
return
# System prompt ConfigMap
prompt_cm = client.V1ConfigMap(
metadata=client.V1ObjectMeta(
name=f"agent-prompt-{agent_id}",
namespace=settings.k8s_namespace,
labels={"app": "roboco-agent", "agent-id": agent_id},
),
data={"system-prompt.md": system_prompt},
)
# MCP config ConfigMap
mcp_cm = client.V1ConfigMap(
metadata=client.V1ObjectMeta(
name=f"agent-mcp-{agent_id}",
namespace=settings.k8s_namespace,
labels={"app": "roboco-agent", "agent-id": agent_id},
),
data={"mcp-config.json": json.dumps(mcp_config, indent=2)},
)
for cm in [prompt_cm, mcp_cm]:
try:
# Try to create, update if exists
try:
self._k8s_core_v1.create_namespaced_config_map(
namespace=settings.k8s_namespace,
body=cm,
)
except ApiException as e:
# 409 = conflict (already exists), so update instead
if e.status == 409: # noqa: PLR2004
self._k8s_core_v1.replace_namespaced_config_map(
name=cm.metadata.name,
namespace=settings.k8s_namespace,
body=cm,
)
else:
raise
except ApiException as e:
logger.error(
"K8s: failed to create configmap",
name=cm.metadata.name,
error=e,
)
raise
async def _generate_mcp_config(self, agent_id: str) -> Path:
"""Generate role-aware MCP config for an agent.
@@ -1165,7 +790,7 @@ class AgentOrchestrator:
# =========================================================================
async def stop_agent(self, agent_id: str, graceful: bool = True) -> None:
"""Stop an agent container or K8s Job."""
"""Stop an agent container."""
async with self._lock:
if agent_id not in self._instances:
return
@@ -1174,12 +799,6 @@ class AgentOrchestrator:
if instance.container_id:
instance.state = AgentState.STOPPING
if settings.k8s_enabled and self._k8s_initialized:
# K8s mode: Delete the Job
await self._stop_k8s_agent(agent_id)
else:
# Docker mode: Stop and remove container
container_name = f"roboco-agent-{agent_id}"
if graceful:
@@ -1211,9 +830,7 @@ class AgentOrchestrator:
instance.state = AgentState.OFFLINE
instance.container_id = None
logger.info(
"Agent stopped", agent_id=agent_id, k8s_enabled=settings.k8s_enabled
)
logger.info("Agent stopped", agent_id=agent_id)
# =========================================================================
# WAITING STATE MANAGEMENT
@@ -1369,13 +986,7 @@ Start by:
if instance.container_id is None:
continue
# Check if agent is still running
if settings.k8s_enabled and self._k8s_initialized:
# K8s mode: Check Job status
k8s_state = await self._get_k8s_agent_status(agent_id)
is_running = k8s_state in (AgentState.ACTIVE, AgentState.STARTING)
else:
# Docker mode: Check container status
# Check if container is still running
container_name = f"roboco-agent-{agent_id}"
proc = await asyncio.create_subprocess_exec(
"docker",
@@ -1387,15 +998,15 @@ Start by:
stderr=asyncio.subprocess.DEVNULL,
)
stdout, _ = await proc.communicate()
is_running = stdout.decode().strip() == "true"
if not is_running:
cid = instance.container_id[:12] if instance.container_id else None
logger.warning(
"Agent container/job stopped",
"Agent container stopped",
agent_id=agent_id,
container_id=cid,
k8s_enabled=settings.k8s_enabled,
)
instance.state = AgentState.OFFLINE
instance.error_count += 1
roboco/services/a2a.py
-480
View File
@@ -1,480 +0,0 @@
"""
A2A (Agent-to-Agent) Protocol Service
Provides business logic for A2A protocol operations including:
- Agent discovery and card generation
- Task lifecycle management via A2A semantics
- Message handling and routing
"""
from uuid import UUID
import structlog
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from roboco.config import settings
from roboco.db.tables import AgentTable, TaskTable
from roboco.models.a2a import (
A2AArtifact,
A2AMessage,
A2ATask,
A2ATaskStatus,
AgentCapabilities,
AgentCard,
AgentProvider,
AgentSkill,
SecurityScheme,
TextPart,
task_status_to_a2a_state,
)
from roboco.models.base import TaskStatus, Team
logger = structlog.get_logger()
class A2AService:
"""
Service layer for A2A protocol operations.
Provides methods for:
- Building Agent Cards for discovery
- Converting between RoboCo tasks and A2A tasks
- Processing A2A messages
"""
def __init__(self, session: AsyncSession):
"""Initialize with database session."""
self.session = session
@staticmethod
def get_service_endpoint() -> str:
"""Build service endpoint URL from settings."""
connect_host = "127.0.0.1" if settings.host == "0.0.0.0" else settings.host
return f"http://{connect_host}:{settings.port}"
@staticmethod
def build_system_agent_card() -> AgentCard:
"""
Build the system-level Agent Card for RoboCo.
This card represents the entire RoboCo system and is served
at /.well-known/agent.json
"""
return AgentCard(
id="roboco-system",
name="RoboCo System",
description=(
"RoboCo is an AI Agentic Company - a virtual organization of "
"AI agents designed to operate as a complete software "
"development workforce."
),
provider=AgentProvider(
organization="RoboCo",
url="https://github.com/roboco",
),
protocol_version="1.0",
service_endpoint=f"{A2AService.get_service_endpoint()}/api/v1/a2a",
version=settings.app_version,
capabilities=AgentCapabilities(
streaming=True,
push_notifications=False,
state_transition_history=True,
),
default_input_modes=["text/plain", "application/json"],
default_output_modes=["text/plain", "application/json"],
skills=[
AgentSkill(
id="software-development",
name="Software Development",
description="Full-stack software development with AI agents",
tags=["development", "coding", "qa", "documentation"],
),
AgentSkill(
id="task-management",
name="Task Management",
description="Create and manage development tasks",
tags=["tasks", "kanban", "planning"],
),
AgentSkill(
id="code-review",
name="Code Review",
description="Review and quality assurance of code",
tags=["qa", "review", "testing"],
),
],
documentation_url="https://github.com/roboco/docs",
security_schemes={
"bearerAuth": SecurityScheme(type="http", scheme="bearer"),
},
security=[{"bearerAuth": []}],
)
async def build_agent_card(self, agent_id: str) -> AgentCard | None:
"""
Build an Agent Card for a specific agent.
Args:
agent_id: Either a UUID string or agent slug
Returns:
AgentCard for the agent, or None if not found
"""
# Try to parse as UUID first
try:
uuid = UUID(agent_id)
result = await self.session.execute(
select(AgentTable).where(AgentTable.id == uuid)
)
except ValueError:
# Not a UUID, try slug lookup
result = await self.session.execute(
select(AgentTable).where(AgentTable.slug == agent_id)
)
agent = result.scalar_one_or_none()
if agent is None:
return None
return self._agent_to_card(agent)
def _agent_to_card(self, agent: AgentTable) -> AgentCard:
"""Convert an AgentTable row to an AgentCard."""
agent_id = str(agent.id)
agent_slug = agent.slug
# Map role to skills
role_skills: dict[str, list[AgentSkill]] = {
"developer": [
AgentSkill(
id="coding",
name="Code Development",
description="Write and implement code",
tags=["development", "coding"],
),
AgentSkill(
id="debugging",
name="Debugging",
description="Debug and fix code issues",
tags=["debugging", "troubleshooting"],
),
],
"qa": [
AgentSkill(
id="testing",
name="Testing",
description="Test code and verify quality",
tags=["qa", "testing"],
),
AgentSkill(
id="review",
name="Code Review",
description="Review code for quality and issues",
tags=["qa", "review"],
),
],
"documenter": [
AgentSkill(
id="documentation",
name="Documentation",
description="Write technical documentation",
tags=["documentation", "writing"],
),
],
"cell_pm": [
AgentSkill(
id="coordination",
name="Task Coordination",
description="Coordinate tasks within the cell",
tags=["management", "coordination"],
),
],
"main_pm": [
AgentSkill(
id="planning",
name="Project Planning",
description="Plan and coordinate across cells",
tags=["management", "planning"],
),
],
}
skills = role_skills.get(agent.role, [])
return AgentCard(
id=agent_id,
name=agent.name,
description=f"{agent.name} - {agent.role} agent in RoboCo",
provider=AgentProvider(
organization="RoboCo",
url="https://github.com/roboco",
),
protocol_version="1.0",
service_endpoint=f"{self.get_service_endpoint()}/api/v1/a2a",
version=settings.app_version,
capabilities=AgentCapabilities(
streaming=True,
push_notifications=False,
state_transition_history=True,
),
default_input_modes=["text/plain", "application/json"],
default_output_modes=["text/plain", "application/json"],
skills=skills,
metadata={
"slug": agent_slug,
"role": agent.role,
"team": agent.team,
},
security_schemes={
"bearerAuth": SecurityScheme(type="http", scheme="bearer"),
},
security=[{"bearerAuth": []}],
)
def task_to_a2a(self, task: TaskTable) -> A2ATask:
"""
Convert a RoboCo TaskTable to A2A Task.
This is the canonical conversion that maintains semantic
mapping between RoboCo's internal task model and A2A.
"""
task_id = str(task.id)
# Get status value as string
if hasattr(task.status, "value"):
status_value = task.status.value
else:
status_value = str(task.status)
a2a_state = task_status_to_a2a_state(status_value)
# Build status message from dev_notes if present
status_message = None
if task.dev_notes:
status_message = A2AMessage(
role="agent",
parts=[TextPart(text=task.dev_notes)],
task_id=task_id,
)
a2a_status = A2ATaskStatus(
state=a2a_state,
message=status_message,
timestamp=task.updated_at or task.created_at,
)
# Build artifacts from task outputs (future: populate from outputs)
artifacts: list[A2AArtifact] = []
# Build metadata
metadata: dict[str, str | int] = {
"roboco_status": status_value,
"priority": task.priority,
"team": str(task.team),
}
if task.assigned_to:
metadata["assigned_to"] = str(task.assigned_to)
if task.parent_task_id:
metadata["parent_task_id"] = str(task.parent_task_id)
return A2ATask(
id=task_id,
context_id=task_id,
status=a2a_status,
artifacts=artifacts,
history=[],
metadata=metadata,
)
async def get_task(self, task_id: str) -> A2ATask | None:
"""
Get a task by ID and return as A2A Task.
Args:
task_id: Task UUID string
Returns:
A2ATask or None if not found
"""
try:
task_uuid = UUID(task_id)
except ValueError:
return None
result = await self.session.execute(
select(TaskTable).where(TaskTable.id == task_uuid)
)
task = result.scalar_one_or_none()
if task is None:
return None
return self.task_to_a2a(task)
async def list_tasks(
self,
page_size: int = 20,
offset: int = 0,
order_by: str | None = None,
) -> tuple[list[A2ATask], bool]:
"""
List tasks with pagination.
Args:
page_size: Number of results to return
offset: Starting offset
order_by: Sort order ("created_at desc" or "created_at asc")
Returns:
Tuple of (tasks, has_more)
"""
query = select(TaskTable)
# Apply ordering
if order_by == "created_at asc":
query = query.order_by(TaskTable.created_at.asc())
else:
query = query.order_by(TaskTable.created_at.desc())
# Apply pagination (fetch one extra to detect more)
query = query.offset(offset).limit(page_size + 1)
result = await self.session.execute(query)
tasks = list(result.scalars().all())
has_more = len(tasks) > page_size
if has_more:
tasks = tasks[:page_size]
return [self.task_to_a2a(t) for t in tasks], has_more
async def create_task_from_message(
self,
title: str,
description: str,
created_by: UUID,
team: Team = Team.BACKEND,
) -> A2ATask:
"""
Create a new task from an A2A message.
Args:
title: Task title
description: Task description
created_by: Agent ID creating the task
team: Team assignment
Returns:
Created A2ATask
"""
task = TaskTable(
title=title,
description=description,
acceptance_criteria=["Task completed as specified"],
status=TaskStatus.PENDING,
priority=5,
team=team,
created_by=created_by,
)
self.session.add(task)
await self.session.flush()
await self.session.refresh(task)
logger.info(
"Created task from A2A message",
task_id=str(task.id),
title=title,
)
return self.task_to_a2a(task)
async def cancel_task(self, task_id: str, reason: str | None = None) -> A2ATask:
"""
Cancel a task.
Args:
task_id: Task UUID string
reason: Optional cancellation reason
Returns:
Updated A2ATask
Raises:
ValueError: If task not found or already in terminal state
"""
try:
task_uuid = UUID(task_id)
except ValueError as e:
raise ValueError(f"Invalid task ID: {task_id}") from e
result = await self.session.execute(
select(TaskTable).where(TaskTable.id == task_uuid)
)
task = result.scalar_one_or_none()
if task is None:
raise ValueError(f"Task not found: {task_id}")
# Check if cancellable
if hasattr(task.status, "value"):
status_value = task.status.value
else:
status_value = str(task.status)
if status_value in ["completed", "cancelled"]:
raise ValueError(f"Task already in terminal state: {status_value}")
# Cancel the task
task.status = TaskStatus.CANCELLED
if reason:
reason_text = f"Cancellation reason: {reason}"
if task.dev_notes:
task.dev_notes = f"{task.dev_notes}\n\n{reason_text}"
else:
task.dev_notes = reason_text
await self.session.flush()
await self.session.refresh(task)
logger.info("Cancelled task via A2A", task_id=task_id, reason=reason)
return self.task_to_a2a(task)
async def discover_agents(
self,
role: str | None = None,
team: str | None = None,
skill_tag: str | None = None,
) -> list[AgentCard]:
"""
Discover agents matching criteria.
Args:
role: Filter by agent role
team: Filter by team
skill_tag: Filter by skill tag (future)
Returns:
List of matching AgentCards
"""
query = select(AgentTable)
if role:
query = query.where(AgentTable.role == role)
if team:
query = query.where(AgentTable.team == team)
result = await self.session.execute(query)
agents = result.scalars().all()
cards = [self._agent_to_card(agent) for agent in agents]
# Filter by skill tag if specified
if skill_tag:
cards = [
card
for card in cards
if any(skill_tag in skill.tags for skill in card.skills)
]
return cards
uv.lock
Generated
+10 -110
View File
@@ -401,15 +401,6 @@ filecache = [
{ name = "filelock" },
]
[[package]]
name = "cachetools"
version = "6.2.4"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/bc/1d/ede8680603f6016887c062a2cf4fc8fdba905866a3ab8831aa8aa651320c/cachetools-6.2.4.tar.gz", hash = "sha256:82c5c05585e70b6ba2d3ae09ea60b79548872185d2f24ae1f2709d37299fd607", size = 31731, upload-time = "2025-12-15T18:24:53.744Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/2c/fc/1d7b80d0eb7b714984ce40efc78859c022cd930e402f599d8ca9e39c78a4/cachetools-6.2.4-py3-none-any.whl", hash = "sha256:69a7a52634fed8b8bf6e24a050fb60bff1c9bd8f6d24572b99c32d4e71e62a51", size = 11551, upload-time = "2025-12-15T18:24:52.332Z" },
]
[[package]]
name = "certifi"
version = "2025.11.12"
@@ -773,15 +764,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/56/26/035d1c308882514a1e6ddca27f9d3e570d67a0e293e7b4d910a70c8fe32b/dparse-0.6.4-py3-none-any.whl", hash = "sha256:fbab4d50d54d0e739fbb4dedfc3d92771003a5b9aa8545ca7a7045e3b174af57", size = 11925, upload-time = "2024-11-08T16:52:03.844Z" },
]
[[package]]
name = "durationpy"
version = "0.10"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/9d/a4/e44218c2b394e31a6dd0d6b095c4e1f32d0be54c2a4b250032d717647bab/durationpy-0.10.tar.gz", hash = "sha256:1fa6893409a6e739c9c72334fc65cca1f355dbdd93405d30f726deb5bde42fba", size = 3335, upload-time = "2025-05-17T13:52:37.26Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/b0/0d/9feae160378a3553fa9a339b0e9c1a048e147a4127210e286ef18b730f03/durationpy-0.10-py3-none-any.whl", hash = "sha256:3b41e1b601234296b4fb368338fdcd3e13e0b4fb5b67345948f4f2bf9868b286", size = 3922, upload-time = "2025-05-17T13:52:36.463Z" },
]
[[package]]
name = "ecdsa"
version = "0.19.1"
@@ -933,20 +915,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/27/e8/68e274b2a30e1fdfd25bdc27194382be3f233929c8f727c0440d58ac074f/glom-22.1.0-py2.py3-none-any.whl", hash = "sha256:5339da206bf3532e01a83a35aca202960ea885156986d190574b779598e9e772", size = 100687, upload-time = "2022-01-24T09:34:02.391Z" },
]
[[package]]
name = "google-auth"
version = "2.45.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "cachetools" },
{ name = "pyasn1-modules" },
{ name = "rsa" },
]
sdist = { url = "https://files.pythonhosted.org/packages/e5/00/3c794502a8b892c404b2dea5b3650eb21bfc7069612fbfd15c7f17c1cb0d/google_auth-2.45.0.tar.gz", hash = "sha256:90d3f41b6b72ea72dd9811e765699ee491ab24139f34ebf1ca2b9cc0c38708f3", size = 320708, upload-time = "2025-12-15T22:58:42.889Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c6/97/451d55e05487a5cd6279a01a7e34921858b16f7dc8aa38a2c684743cd2b3/google_auth-2.45.0-py2.py3-none-any.whl", hash = "sha256:82344e86dc00410ef5382d99be677c6043d72e502b625aa4f4afa0bdacca0f36", size = 233312, upload-time = "2025-12-15T22:58:40.777Z" },
]
[[package]]
name = "googleapis-common-protos"
version = "1.72.0"
@@ -1360,27 +1328,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/41/45/1a4ed80516f02155c51f51e8cedb3c1902296743db0bbc66608a0db2814f/jsonschema_specifications-2025.9.1-py3-none-any.whl", hash = "sha256:98802fee3a11ee76ecaca44429fda8a41bff98b00a0f2838151b113f210cc6fe", size = 18437, upload-time = "2025-09-08T01:34:57.871Z" },
]
[[package]]
name = "kubernetes"
version = "34.1.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "certifi" },
{ name = "durationpy" },
{ name = "google-auth" },
{ name = "python-dateutil" },
{ name = "pyyaml" },
{ name = "requests" },
{ name = "requests-oauthlib" },
{ name = "six" },
{ name = "urllib3" },
{ name = "websocket-client" },
]
sdist = { url = "https://files.pythonhosted.org/packages/ef/55/3f880ef65f559cbed44a9aa20d3bdbc219a2c3a3bac4a30a513029b03ee9/kubernetes-34.1.0.tar.gz", hash = "sha256:8fe8edb0b5d290a2f3ac06596b23f87c658977d46b5f8df9d0f4ea83d0003912", size = 1083771, upload-time = "2025-09-29T20:23:49.283Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/ca/ec/65f7d563aa4a62dd58777e8f6aa882f15db53b14eb29aba0c28a20f7eb26/kubernetes-34.1.0-py2.py3-none-any.whl", hash = "sha256:bffba2272534e224e6a7a74d582deb0b545b7c9879d2cd9e4aae9481d1f2cc2a", size = 2008380, upload-time = "2025-09-29T20:23:47.684Z" },
]
[[package]]
name = "lance-namespace"
version = "0.4.0"
@@ -2120,7 +2067,7 @@ name = "nvidia-cudnn-cu12"
version = "9.10.2.21"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "nvidia-cublas-cu12", marker = "sys_platform != 'win32'" },
{ name = "nvidia-cublas-cu12" },
]
wheels = [
{ url = "https://files.pythonhosted.org/packages/ba/51/e123d997aa098c61d029f76663dedbfb9bc8dcf8c60cbd6adbe42f76d049/nvidia_cudnn_cu12-9.10.2.21-py3-none-manylinux_2_27_x86_64.whl", hash = "sha256:949452be657fa16687d0930933f032835951ef0892b37d2d53824d1a84dc97a8", size = 706758467, upload-time = "2025-06-06T21:54:08.597Z" },
@@ -2131,7 +2078,7 @@ name = "nvidia-cufft-cu12"
version = "11.3.3.83"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "nvidia-nvjitlink-cu12", marker = "sys_platform != 'win32'" },
{ name = "nvidia-nvjitlink-cu12" },
]
wheels = [
{ url = "https://files.pythonhosted.org/packages/1f/13/ee4e00f30e676b66ae65b4f08cb5bcbb8392c03f54f2d5413ea99a5d1c80/nvidia_cufft_cu12-11.3.3.83-py3-none-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:4d2dd21ec0b88cf61b62e6b43564355e5222e4a3fb394cac0db101f2dd0d4f74", size = 193118695, upload-time = "2025-03-07T01:45:27.821Z" },
@@ -2158,9 +2105,9 @@ name = "nvidia-cusolver-cu12"
version = "11.7.3.90"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "nvidia-cublas-cu12", marker = "sys_platform != 'win32'" },
{ name = "nvidia-cusparse-cu12", marker = "sys_platform != 'win32'" },
{ name = "nvidia-nvjitlink-cu12", marker = "sys_platform != 'win32'" },
{ name = "nvidia-cublas-cu12" },
{ name = "nvidia-cusparse-cu12" },
{ name = "nvidia-nvjitlink-cu12" },
]
wheels = [
{ url = "https://files.pythonhosted.org/packages/85/48/9a13d2975803e8cf2777d5ed57b87a0b6ca2cc795f9a4f59796a910bfb80/nvidia_cusolver_cu12-11.7.3.90-py3-none-manylinux_2_27_x86_64.whl", hash = "sha256:4376c11ad263152bd50ea295c05370360776f8c3427b30991df774f9fb26c450", size = 267506905, upload-time = "2025-03-07T01:47:16.273Z" },
@@ -2171,7 +2118,7 @@ name = "nvidia-cusparse-cu12"
version = "12.5.8.93"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "nvidia-nvjitlink-cu12", marker = "sys_platform != 'win32'" },
{ name = "nvidia-nvjitlink-cu12" },
]
wheels = [
{ url = "https://files.pythonhosted.org/packages/c2/f5/e1854cb2f2bcd4280c44736c93550cc300ff4b8c95ebe370d0aa7d2b473d/nvidia_cusparse_cu12-12.5.8.93-py3-none-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:1ec05d76bbbd8b61b06a80e1eaf8cf4959c3d4ce8e711b65ebd0443bb0ebb13b", size = 288216466, upload-time = "2025-03-07T01:48:13.779Z" },
@@ -2217,15 +2164,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/a2/eb/86626c1bbc2edb86323022371c39aa48df6fd8b0a1647bc274577f72e90b/nvidia_nvtx_cu12-12.8.90-py3-none-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:5b17e2001cc0d751a5bc2c6ec6d26ad95913324a4adb86788c944f8ce9ba441f", size = 89954, upload-time = "2025-03-07T01:42:44.131Z" },
]
[[package]]
name = "oauthlib"
version = "3.3.1"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/0b/5f/19930f824ffeb0ad4372da4812c50edbd1434f678c90c2733e1188edfc63/oauthlib-3.3.1.tar.gz", hash = "sha256:0f0f8aa759826a193cf66c12ea1af1637f87b9b4622d46e866952bb022e538c9", size = 185918, upload-time = "2025-06-19T22:48:08.269Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/be/9c/92789c596b8df838baa98fa71844d84283302f7604ed565dafe5a6b5041a/oauthlib-3.3.1-py3-none-any.whl", hash = "sha256:88119c938d2b8fb88561af5f6ee0eec8cc8d552b7bb1f712743136eb7523b7a1", size = 160065, upload-time = "2025-06-19T22:48:06.508Z" },
]
[[package]]
name = "olefile"
version = "0.47"
@@ -2531,7 +2469,7 @@ name = "pexpect"
version = "4.9.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "ptyprocess", marker = "sys_platform != 'win32'" },
{ name = "ptyprocess" },
]
sdist = { url = "https://files.pythonhosted.org/packages/42/92/cc564bf6381ff43ce1f4d06852fc19a2f11d180f23dc32d9588bee2f149d/pexpect-4.9.0.tar.gz", hash = "sha256:ee7d41123f3c9911050ea2c2dac107568dc43b2d3b0c7557a33212c398ead30f", size = 166450, upload-time = "2023-11-25T09:07:26.339Z" }
wheels = [
@@ -2840,18 +2778,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/c8/f1/d6a797abb14f6283c0ddff96bbdd46937f64122b8c925cab503dd37f8214/pyasn1-0.6.1-py3-none-any.whl", hash = "sha256:0d632f46f2ba09143da3a8afe9e33fb6f92fa2320ab7e886e2d0f7672af84629", size = 83135, upload-time = "2024-09-11T16:00:36.122Z" },
]
[[package]]
name = "pyasn1-modules"
version = "0.4.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pyasn1" },
]
sdist = { url = "https://files.pythonhosted.org/packages/e9/e6/78ebbb10a8c8e4b61a59249394a4a594c1a7af95593dc933a349c8d00964/pyasn1_modules-0.4.2.tar.gz", hash = "sha256:677091de870a80aae844b1ca6134f54652fa2c8c5a52aa396440ac3106e941e6", size = 307892, upload-time = "2025-03-28T02:41:22.17Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/47/8d/d529b5d697919ba8c11ad626e835d4039be708a35b0d22de83a269a6682c/pyasn1_modules-0.4.2-py3-none-any.whl", hash = "sha256:29253a9207ce32b64c3ac6600edc75368f98473906e8fd1043bd6b5b1de2c14a", size = 181259, upload-time = "2025-03-28T02:41:19.028Z" },
]
[[package]]
name = "pycparser"
version = "2.23"
@@ -3326,19 +3252,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/1e/db/4254e3eabe8020b458f1a747140d32277ec7a271daf1d235b70dc0b4e6e3/requests-2.32.5-py3-none-any.whl", hash = "sha256:2462f94637a34fd532264295e186976db0f5d453d1cdd31473c85a6a161affb6", size = 64738, upload-time = "2025-08-18T20:46:00.542Z" },
]
[[package]]
name = "requests-oauthlib"
version = "2.0.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "oauthlib" },
{ name = "requests" },
]
sdist = { url = "https://files.pythonhosted.org/packages/42/f2/05f29bc3913aea15eb670be136045bf5c5bbf4b99ecb839da9b422bb2c85/requests-oauthlib-2.0.0.tar.gz", hash = "sha256:b3dffaebd884d8cd778494369603a9e7b58d29111bf6b41bdc2dcd87203af4e9", size = 55650, upload-time = "2024-03-22T20:32:29.939Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/3b/5d/63d4ae3b9daea098d5d6f5da83984853c1bbacd5dc826764b249fe119d24/requests_oauthlib-2.0.0-py2.py3-none-any.whl", hash = "sha256:7dd8a5c40426b779b0868c404bdef9768deccf22749cde15852df527e6269b36", size = 24179, upload-time = "2024-03-22T20:32:28.055Z" },
]
[[package]]
name = "requirements-parser"
version = "0.13.0"
@@ -3376,7 +3289,6 @@ dependencies = [
{ name = "fastapi" },
{ name = "hiredis" },
{ name = "httpx" },
{ name = "kubernetes" },
{ name = "mcp" },
{ name = "openai" },
{ name = "passlib", extra = ["bcrypt"] },
@@ -3388,7 +3300,6 @@ dependencies = [
{ name = "python-toon" },
{ name = "redis" },
{ name = "sqlalchemy", extra = ["asyncio"] },
{ name = "sse-starlette" },
{ name = "structlog" },
{ name = "tenacity" },
{ name = "tiktoken" },
@@ -3447,7 +3358,6 @@ requires-dist = [
{ name = "httpx" },
{ name = "httpx", marker = "extra == 'dev'" },
{ name = "ipython", marker = "extra == 'dev'" },
{ name = "kubernetes" },
{ name = "mcp" },
{ name = "mkdocs", marker = "extra == 'docs'" },
{ name = "mkdocs-material", marker = "extra == 'docs'" },
@@ -3473,7 +3383,6 @@ requires-dist = [
{ name = "safety", marker = "extra == 'dev'" },
{ name = "semgrep", marker = "extra == 'dev'" },
{ name = "sqlalchemy", extras = ["asyncio"] },
{ name = "sse-starlette" },
{ name = "structlog" },
{ name = "tenacity" },
{ name = "tiktoken" },
@@ -4379,11 +4288,11 @@ wheels = [
[[package]]
name = "urllib3"
version = "2.3.0"
version = "2.6.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/aa/63/e53da845320b757bf29ef6a9062f5c669fe997973f966045cb019c3f4b66/urllib3-2.3.0.tar.gz", hash = "sha256:f8c5449b3cf0861679ce7e0503c7b44b5ec981bec0d1d3795a07f1ba96f0204d", size = 307268, upload-time = "2024-12-22T07:47:30.032Z" }
sdist = { url = "https://files.pythonhosted.org/packages/1e/24/a2a2ed9addd907787d7aa0355ba36a6cadf1768b934c652ea78acbd59dcd/urllib3-2.6.2.tar.gz", hash = "sha256:016f9c98bb7e98085cb2b4b17b87d2c702975664e4f060c6532e64d1c1a5e797", size = 432930, upload-time = "2025-12-11T15:56:40.252Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c8/19/4ec628951a74043532ca2cf5d97b7b14863931476d117c471e8e2b1eb39f/urllib3-2.3.0-py3-none-any.whl", hash = "sha256:1cee9ad369867bfdbbb48b7dd50374c0967a0bb7710050facf0dd6911440e3df", size = 128369, upload-time = "2024-12-22T07:47:28.074Z" },
{ url = "https://files.pythonhosted.org/packages/6d/b9/4095b668ea3678bf6a0af005527f39de12fb026516fb3df17495a733b7f8/urllib3-2.6.2-py3-none-any.whl", hash = "sha256:ec21cddfe7724fc7cb4ba4bea7aa8e2ef36f607a4bab81aa6ce42a13dc3f03dd", size = 131182, upload-time = "2025-12-11T15:56:38.584Z" },
]
[[package]]
@@ -4544,15 +4453,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/af/b5/123f13c975e9f27ab9c0770f514345bd406d0e8d3b7a0723af9d43f710af/wcwidth-0.2.14-py2.py3-none-any.whl", hash = "sha256:a7bb560c8aee30f9957e5f9895805edd20602f2d7f720186dfd906e82b4982e1", size = 37286, upload-time = "2025-09-22T16:29:51.641Z" },
]
[[package]]
name = "websocket-client"
version = "1.9.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/2c/41/aa4bf9664e4cda14c3b39865b12251e8e7d239f4cd0e3cc1b6c2ccde25c1/websocket_client-1.9.0.tar.gz", hash = "sha256:9e813624b6eb619999a97dc7958469217c3176312b3a16a4bd1bc7e08a46ec98", size = 70576, upload-time = "2025-10-07T21:16:36.495Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/34/db/b10e48aa8fff7407e67470363eac595018441cf32d5e1001567a7aeba5d2/websocket_client-1.9.0-py3-none-any.whl", hash = "sha256:af248a825037ef591efbf6ed20cc5faa03d3b47b9e5a2230a529eeee1c1fc3ef", size = 82616, upload-time = "2025-10-07T21:16:34.951Z" },
]
[[package]]
name = "websockets"
version = "15.0.1"