Teams Service

The Teams Service enables you to create and manage multi-agent teams that can collaborate on complex tasks. Teams consist of multiple agents working together under a team leader’s coordination.

Quick Start

from kubiya import ControlPlaneClient

client = ControlPlaneClient()

# List all teams
teams = client.teams.list()

# Get a specific team
team = client.teams.get(team_id="team-uuid")

# Create a new team
team_data = {
    "name": "DevOps Team",
    "description": "Team for infrastructure automation",
    "runtime": "default"
}
created_team = client.teams.create(team_data)

Features

Team Management

Create, update, and delete multi-agent teams

Agent Membership

Add and remove agents from teams

Team Execution

Execute tasks with team collaboration

Streaming Responses

Real-time streaming execution results

List Teams

Retrieve all teams in your organization with optional filtering:

# List all teams
teams = client.teams.list()

# List with pagination
teams = client.teams.list(skip=0, limit=50)

# Filter by status
active_teams = client.teams.list(status_filter="active")
inactive_teams = client.teams.list(status_filter="inactive")

Parameters:

skip (int, optional): Number of records to skip for pagination (default: 0)
limit (int, optional): Maximum number of teams to return (default: 100)
status_filter (str, optional): Filter by status: ‘active’, ‘inactive’, ‘archived’, or ‘idle’

Returns: List of team dictionaries with agent information

Get Team

Retrieve a specific team by ID:

team = client.teams.get(team_id="550e8400-e29b-41d4-a716-446655440000")

print(f"Team: {team['name']}")
print(f"Description: {team['description']}")
print(f"Runtime: {team['runtime']}")
print(f"Agents: {len(team.get('agents', []))}")

Parameters:

team_id (str, required): Team UUID

Returns: Dictionary containing team details with associated agents

Create Team

Create a new team with configuration:

team_data = {
    "name": "Infrastructure Team",
    "description": "Manages cloud infrastructure and deployments",
    "runtime": "default",  # or "claude_code"
    "configuration": {
        "max_concurrent_tasks": 5,
        "timeout_seconds": 3600
    },
    "skill_ids": ["skill-uuid-1", "skill-uuid-2"],
    "environment_ids": ["env-uuid-1"]
}

team = client.teams.create(team_data)
print(f"Created team: {team['id']}")

Parameters:

team_data (dict, required): Team configuration
- name (str): Team name
- description (str): Team description
- runtime (str): Runtime type - ‘default’ or ‘claude_code’
- configuration (dict, optional): Team configuration settings
- skill_ids (list, optional): List of skill UUIDs
- skill_configurations (dict, optional): Skill-specific configs
- environment_ids (list, optional): List of environment UUIDs
- execution_environment (dict, optional): Execution environment config

Returns: Dictionary containing created team details

Update Team

Update an existing team (partial update):

# Update team name and description
updated_team = client.teams.update(
    team_id="team-uuid",
    team_data={
        "name": "Updated Team Name",
        "description": "New description"
    }
)

# Update team skills
updated_team = client.teams.update(
    team_id="team-uuid",
    team_data={
        "skill_ids": ["skill-uuid-1", "skill-uuid-2", "skill-uuid-3"]
    }
)

# Change runtime
updated_team = client.teams.update(
    team_id="team-uuid",
    team_data={
        "runtime": "claude_code"
    }
)

Parameters:

team_id (str, required): Team UUID
team_data (dict, required): Fields to update (partial update)
- name (str, optional): Team name
- description (str, optional): Team description
- status (str, optional): Team status
- runtime (str, optional): Runtime type
- configuration (dict, optional): Team configuration
- skill_ids (list, optional): List of skill IDs
- skill_configurations (dict, optional): Skill configs
- environment_ids (list, optional): Environment IDs
- execution_environment (dict, optional): Execution environment config

Returns: Dictionary containing updated team details

Delete Team

Delete a team:

client.teams.delete(team_id="team-uuid")
print("Team deleted successfully")

Parameters:

team_id (str, required): Team UUID

Returns: None (204 No Content on success)

Deleting a team is permanent and cannot be undone. Ensure you have backups of any important team configurations.

Agent Management

Add Agent to Team

Add an agent to a team by setting the agent’s team_id:

result = client.teams.add_agent(
    team_id="team-uuid",
    agent_id="agent-uuid"
)

print(f"Agent added to team: {result['name']}")
print(f"Total agents: {len(result.get('agents', []))}")

Parameters:

team_id (str, required): Team UUID
agent_id (str, required): Agent UUID to add

Returns: Dictionary containing updated team details with agents

Remove Agent from Team

Remove an agent from a team:

result = client.teams.remove_agent(
    team_id="team-uuid",
    agent_id="agent-uuid"
)

print(f"Agent removed from team: {result['name']}")

Parameters:

team_id (str, required): Team UUID
agent_id (str, required): Agent UUID to remove

Returns: Dictionary containing updated team details with agents

Team Execution

Execute Team Task

Execute a task using team collaboration:

execution_data = {
    "prompt": "Deploy the latest version to production",
    "worker_queue_id": "queue-uuid"
}

execution = client.teams.execute(
    team_id="team-uuid",
    execution_data=execution_data
)

print(f"Execution ID: {execution['execution_id']}")
print(f"Workflow ID: {execution['workflow_id']}")
print(f"Status: {execution['status']}")

Parameters:

team_id (str, required): Team UUID
execution_data (dict, required): Execution request
- prompt (str, required): The task prompt
- worker_queue_id (str, required): Worker queue UUID
- system_prompt (str, optional): Optional system prompt
- user_metadata (dict, optional): User attribution metadata

Returns: Dictionary containing execution details

The execution creates a record and starts a Temporal workflow. The actual execution happens asynchronously.

Execute with Streaming

Execute a team task with real-time streaming responses:

execution_data = {
    "prompt": "Analyze the production database performance",
    "stream": True,
    "worker_queue_id": "queue-uuid"
}

# Stream the execution results
for chunk in client.teams.execute_stream(
    team_id="team-uuid",
    execution_data=execution_data
):
    # Process each chunk as it arrives
    print(chunk.decode('utf-8'), end='', flush=True)

Parameters:

team_id (str, required): Team UUID
execution_data (dict, required): Execution request
- prompt (str, required): The task prompt
- stream (bool): Should be True for streaming
- worker_queue_id (str, required): Worker queue UUID
- system_prompt (str, optional): Optional system prompt
- user_metadata (dict, optional): User attribution metadata

Returns: Iterator of response chunks (Server-Sent Events)

Use streaming for long-running tasks to get real-time progress updates as the team works on the task.

Complete Example

Here’s a complete example of creating and using a team:

from kubiya import ControlPlaneClient

client = ControlPlaneClient()

# 1. Create a new team
team_data = {
    "name": "DevOps Automation Team",
    "description": "Handles infrastructure and deployment tasks",
    "runtime": "default",
    "configuration": {
        "max_concurrent_tasks": 3
    }
}

team = client.teams.create(team_data)
print(f"Created team: {team['id']}")

# 2. Add agents to the team
agent_ids = ["agent-1-uuid", "agent-2-uuid", "agent-3-uuid"]

for agent_id in agent_ids:
    result = client.teams.add_agent(
        team_id=team['id'],
        agent_id=agent_id
    )
    print(f"Added agent {agent_id}")

# 3. Execute a task with the team
execution_data = {
    "prompt": "Check the health of all production services",
    "worker_queue_id": "production-queue-uuid",
    "system_prompt": "Coordinate with team members to check all services"
}

print("\nExecuting team task...")
for chunk in client.teams.execute_stream(
    team_id=team['id'],
    execution_data=execution_data
):
    print(chunk.decode('utf-8'), end='', flush=True)

# 4. Get updated team info
updated_team = client.teams.get(team_id=team['id'])
print(f"\n\nTeam has {len(updated_team.get('agents', []))} agents")

Error Handling

Handle team-related errors:

from kubiya.resources.exceptions import TeamError

try:
    team = client.teams.get(team_id="invalid-id")
except TeamError as e:
    print(f"Team error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

Best Practices

Team Organization

Specialized Teams: Create teams for specific domains (e.g., infrastructure, security, data)
Right-Sized Teams: Keep teams focused with 3-7 agents for optimal coordination
Clear Naming: Use descriptive team names that indicate their purpose

Agent Assignment

Complementary Skills: Add agents with complementary skill sets
Load Distribution: Balance workload across team members
Skill Overlap: Include some skill overlap for redundancy

Execution

Use Streaming: For long-running tasks, use execute_stream() for real-time feedback
Worker Queues: Assign teams to appropriate worker queues for execution isolation
System Prompts: Provide clear coordination instructions in system prompts

Team Lifecycle

Regular Reviews: Periodically review team composition and effectiveness
Archive Inactive: Archive teams that are no longer needed instead of deleting
Version Control: Keep track of team configuration changes

API Reference

Method	Endpoint Pattern	Description
`list()`	GET /teams	List all teams
`get()`	GET /teams/	Get team by ID
`create()`	POST /teams	Create new team
`update()`	PATCH /teams/	Update team
`delete()`	DELETE /teams/	Delete team
`add_agent()`	POST /teams//agents/	Add agent to team
`remove_agent()`	DELETE /teams//agents/	Remove agent
`execute()`	POST /teams//execute	Execute task
`execute_stream()`	POST /teams//execute/stream	Execute with streaming

Next Steps

Agents Service

Manage individual agents

Projects Service

Organize teams in projects

Skills Service

Manage team skills

Workers Service

Configure worker queues

Getting Started

Control Plane Client

Context Graph & Memory

Advanced

Resources

Quick Start

Features

Team Management

Agent Membership

Team Execution

Streaming Responses

List Teams

Get Team

Create Team

Update Team

Delete Team

Agent Management

Add Agent to Team

Remove Agent from Team

Team Execution

Execute Team Task

Execute with Streaming

Complete Example

Error Handling

Best Practices

Team Organization

Agent Assignment

Execution

Team Lifecycle

API Reference

Next Steps

Agents Service

Projects Service

Skills Service

Workers Service

Getting Started

Control Plane Client

Context Graph & Memory

Advanced

Resources

Documentation Index

​Quick Start

​Features

Team Management

Agent Membership

Team Execution

Streaming Responses

​List Teams

​Get Team

​Create Team

​Update Team

​Delete Team

​Agent Management

​Add Agent to Team

​Remove Agent from Team

​Team Execution

​Execute Team Task

​Execute with Streaming

​Complete Example

​Error Handling

​Best Practices

​Team Organization

​Agent Assignment

​Execution

​Team Lifecycle

​API Reference

​Next Steps

Agents Service

Projects Service

Skills Service

Workers Service

Quick Start

Features

List Teams

Get Team

Create Team

Update Team

Delete Team

Agent Management

Add Agent to Team

Remove Agent from Team

Team Execution

Execute Team Task

Execute with Streaming

Complete Example

Error Handling

Best Practices

Team Organization

Agent Assignment

Execution

Team Lifecycle

API Reference

Next Steps