Agents API¶
This section documents the core Agent classes, including the main Agent class, agent composition tools, and agent team functionality.
The main module of the Agentle framework for creating and managing AI agents. |
|
Agent Pipeline Module for the Agentle Framework |
|
Agent Team Module for the Agentle Framework |
|
Context management module for Agentle agents. |
|
Configuration module for Agentle agents. |
|
Module for representing and managing agent execution results. |
|
Agent-to-Agent (A2A) Interface |
Agent¶
The main module of the Agentle framework for creating and managing AI agents.
This module contains the definition of the Agent class, which is the central component of the Agentle framework. It allows you to create intelligent agents capable of processing different types of input, using external tools, and generating structured responses. The Agent facilitates integration with different AI model providers and supports a wide variety of input formats.
Basic example: ```python from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider from agentle.agents.agent import Agent
- weather_agent = Agent(
generation_provider=GoogleGenerationProvider(), model=”gemini-2.0-flash”, instructions=”You are a weather agent that can answer questions about the weather.”, tools=[get_weather],
)
output = weather_agent.run(“Hello. What is the weather in Tokyo?”) ```
- class Agent(*, uid=<factory>, name='Agent', description='An AI agent', url='in-memory', static_knowledge=<factory>, document_parser=None, document_cache_store=None, generation_provider, file_visual_description_provider=None, file_audio_description_provider=None, version='0.0.1', endpoint=None, documentationUrl=None, capabilities=<factory>, authentication=<factory>, defaultInputModes=<factory>, defaultOutputModes=<factory>, skills=<factory>, model=None, instructions='You are a helpful assistant.', response_schema=None, mcp_servers=<factory>, tools=<factory>, config=<factory>, debug=False, suspension_manager=None, speech_to_text_provider=None)[source]¶
Bases:
BaseModel,GenericThe main class of the Agentle framework that represents an intelligent agent.
An Agent is an entity that can process various types of input, perform tasks using tools, and generate responses that can be structured. It encapsulates all the logic needed to interact with AI models, manage context, call external tools, and format responses.
The Agent class is generic and supports structured response types through the T_Schema type parameter, which can be a Pydantic class to define the expected output structure.
- Parameters:
uid (UUID)
name (str)
description (str)
url (str)
document_parser (DocumentParser | None)
document_cache_store (DocumentCacheStore | None)
generation_provider (GenerationProvider)
file_visual_description_provider (GenerationProvider | None)
file_audio_description_provider (GenerationProvider | None)
version (str)
endpoint (str | None)
documentationUrl (str | None)
capabilities (Capabilities)
authentication (Authentication)
defaultInputModes (Sequence[Annotated[str | _MimeType, BeforeValidator(func=~rsb.models.mimetype._validate_mime_type, json_schema_input_type=PydanticUndefined)]])
defaultOutputModes (Sequence[Annotated[str | _MimeType, BeforeValidator(func=~rsb.models.mimetype._validate_mime_type, json_schema_input_type=PydanticUndefined)]])
skills (Sequence[AgentSkill])
model (str | Literal['category_nano', 'category_mini', 'category_standard', 'category_pro', 'category_flagship', 'category_reasoning', 'category_vision', 'category_coding', 'category_instruct', 'category_nano_experimental', 'category_mini_experimental', 'category_standard_experimental', 'category_pro_experimental', 'category_flagship_experimental', 'category_reasoning_experimental', 'category_vision_experimental', 'category_coding_experimental', 'category_instruct_experimental'] | ~collections.abc.Callable[[...], str] | None)
instructions (str | Prompt | Callable[[], str] | Sequence[str])
response_schema (type[T_Schema] | None)
mcp_servers (Sequence[MCPServerProtocol])
config (AgentConfig | AgentConfigDict)
debug (bool)
suspension_manager (SuspensionManager | None)
speech_to_text_provider (SpeechToTextProvider | None)
- description¶
Description of the agent, used for communication with users and other agents.
- Type:
- generation_provider¶
Generation provider used by the agent.
- Type:
agentle.generations.providers.base.generation_provider.GenerationProvider
- capabilities¶
Optional capabilities supported by the agent.
- Type:
agentle.agents.a2a.models.capabilities.Capabilities
- authentication¶
Authentication requirements for the agent.
- Type:
agentle.agents.a2a.models.authentication.Authentication
- defaultInputModes¶
Input interaction modes supported by the agent.
- Type:
collections.abc.Sequence[str | rsb.models.mimetype._MimeType]
- defaultOutputModes¶
Output interaction modes supported by the agent.
- Type:
collections.abc.Sequence[str | rsb.models.mimetype._MimeType]
- skills¶
Skills that the agent can perform.
- Type:
collections.abc.Sequence[agentle.agents.a2a.models.agent_skill.AgentSkill]
- model¶
Model to be used by the agent’s service provider.
- Type:
str | Literal[‘category_nano’, ‘category_mini’, ‘category_standard’, ‘category_pro’, ‘category_flagship’, ‘category_reasoning’, ‘category_vision’, ‘category_coding’, ‘category_instruct’, ‘category_nano_experimental’, ‘category_mini_experimental’, ‘category_standard_experimental’, ‘category_pro_experimental’, ‘category_flagship_experimental’, ‘category_reasoning_experimental’, ‘category_vision_experimental’, ‘category_coding_experimental’, ‘category_instruct_experimental’] | collections.abc.Callable[[…], str] | None
- instructions¶
Instructions for the agent.
- Type:
str | agentle.prompts.models.prompt.Prompt | collections.abc.Callable[[], str] | collections.abc.Sequence[str]
- mcp_servers¶
MCP servers to be used by the agent.
- tools¶
Tools to be used by the agent.
- Type:
collections.abc.Sequence[agentle.generations.tools.tool.Tool[Any] | collections.abc.Callable[[…], object]]
- config¶
Configuration for the agent.
- Type:
agentle.agents.agent_config.AgentConfig | agentle.agents.agent_config_dict.AgentConfigDict
Example
```python from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider from agentle.agents.agent import Agent
# Define a simple tool def get_weather(location: str) -> str:
return f”The weather in {location} is sunny.”
# Create a weather agent weather_agent = Agent(
generation_provider=GoogleGenerationProvider(), model=”gemini-2.0-flash”, instructions=”You are a weather agent that can answer questions about the weather.”, tools=[get_weather],
)
# Run the agent output = weather_agent.run(“What is the weather in London?”) ```
- description: str¶
A human-readable description of the agent. Used to assist users and other agents in understanding what the agent can do. (e.g. “Agent that helps users with recipes and cooking.”)
- static_knowledge: Sequence[StaticKnowledge | str]¶
Static knowledge to be used by the agent. This will be used to enrich the agent’s knowledge base. This will be FULLY (entire document) indexed to the conversation. This can be any url or a local file path.
You can provide a cache duration (in seconds) to cache the parsed content for subsequent calls. Example: ```python agent = Agent(
- static_knowledge=[
StaticKnowledge(content=”https://example.com/data.pdf”, cache=3600), # Cache for 1 hour StaticKnowledge(content=”local_file.txt”, cache=”infinite”), # Cache indefinitely “raw text knowledge” # No caching (default)
]
)¶
- document_parser: DocumentParser | None¶
A document parser to be used by the agent. This will be used to parse the static knowledge documents, if provided.
- document_cache_store: DocumentCacheStore | None¶
A cache store to be used by the agent for caching parsed documents. If None, a default InMemoryDocumentCacheStore will be used.
Example: ```python from agentle.parsing.cache import InMemoryDocumentCacheStore, RedisCacheStore
# Use in-memory cache (default) agent = Agent(document_cache_store=InMemoryDocumentCacheStore())
# Use Redis cache for distributed environments agent = Agent(document_cache_store=RedisCacheStore(redis_url=”redis://localhost:6379/0”)) ```
- generation_provider: GenerationProvider¶
The service provider of the agent
- file_visual_description_provider: GenerationProvider | None¶
The service provider of the agent for visual description.
- file_audio_description_provider: GenerationProvider | None¶
The service provider of the agent for audio description.
- capabilities: Capabilities¶
Optional capabilities supported by the agent.
- authentication: Authentication¶
Authentication requirements for the agent. Intended to match OpenAPI authentication structure.
- defaultInputModes: Sequence[MimeType]¶
The set of interaction modes that the agent supports across all skills. This can be overridden per-skill.
- defaultOutputModes: Sequence[MimeType]¶
The set of interaction modes that the agent supports across all skills. This can be overridden per-skill.
- skills: Sequence[AgentSkill]¶
Skills are a unit of capability that an agent can perform.
- model: str | ModelKind | Callable[..., str] | None¶
The model to use for the agent’s service provider.
- instructions: str | Prompt | Callable[[], str] | Sequence[str]¶
The instructions to use for the agent.
- mcp_servers: Sequence[MCPServerProtocol]¶
The MCP servers to use for the agent.
- config: AgentConfig | AgentConfigDict¶
The configuration for the agent.
- suspension_manager: SuspensionManager | None¶
The suspension manager to use for Human-in-the-Loop workflows. If None, uses the default global suspension manager.
- speech_to_text_provider: SpeechToTextProvider | None¶
The transcription provider to use for speech-to-text.
- model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- property agent_config: AgentConfig¶
- property resolved_model: str | Literal['category_nano', 'category_mini', 'category_standard', 'category_pro', 'category_flagship', 'category_reasoning', 'category_vision', 'category_coding', 'category_instruct', 'category_nano_experimental', 'category_mini_experimental', 'category_standard_experimental', 'category_pro_experimental', 'category_flagship_experimental', 'category_reasoning_experimental', 'category_vision_experimental', 'category_coding_experimental', 'category_instruct_experimental'] | None¶
- classmethod from_agent_card(agent_card)[source]¶
Creates an Agent instance from an A2A agent card.
This method parses an agent card dictionary and creates an Agent instance with the appropriate attributes. It maps the provider organization to a generation provider class if available.
- Parameters:
agent_card (Mapping[str, Any]) – A dictionary representing an A2A agent card
- Returns:
A new Agent instance based on the agent card
- Return type:
Agent[Any]
- Raises:
KeyError – If a required field is missing from the agent card
ValueError – If the provider organization is specified but not supported
Example
```python # Load an agent card from a file with open(“agent_card.json”, “r”) as f:
agent_card = json.load(f)
# Create an agent from the card agent = Agent.from_agent_card(agent_card) ```
- to_agent_card()[source]¶
Generates an A2A agent card from this Agent instance.
This method creates a dictionary representation of the agent in the A2A agent card format, including all relevant attributes such as name, description, capabilities, authentication, and skills.
Example
```python # Create an agent agent = Agent(
name=”Weather Agent”, description=”An agent that provides weather information”, generation_provider=GoogleGenerationProvider(), skills=[
- AgentSkill(
name=”Get Weather”, description=”Gets the current weather for a location”, tags=[“weather”, “forecast”]
)
]
)
# Generate an agent card agent_card = agent.to_agent_card()
# Save the agent card to a file with open(“agent_card.json”, “w”) as f:
json.dump(agent_card, f, indent=2)
- has_tools()[source]¶
Checks if this agent has configured tools.
- Returns:
True if the agent has tools, False otherwise.
- Return type:
- start_mcp_servers()[source]¶
Context manager to connect and clean up MCP servers.
This context manager ensures that all MCP servers are connected before the code block is executed and cleaned up after completion, even in case of exceptions.
- Yields:
None – Does not return a value, just manages the context.
- Return type:
Generator[None, None, None]
Example
```python async with agent.start_mcp_servers():
# Operations that require connection to MCP servers result = await agent.run_async(“Query to server”)
- start_mcp_servers_async()[source]¶
Asynchronous context manager to connect and clean up MCP servers.
This method ensures that all MCP servers are connected before the code block is executed and cleaned up after completion, even in case of exceptions.
- Yields:
None – Does not return a value, just manages the context.
- Return type:
AsyncGenerator[None, None]
Example
```python async with agent.start_mcp_servers():
# Operations that require connection to MCP servers result = await agent.run_async(“Query to server”)
- run(input, *, timeout=None, trace_params=None)[source]¶
Runs the agent synchronously with the provided input.
This method is a synchronous wrapper for run_async, allowing easy use in synchronous contexts.
- Parameters:
input (AgentInput | Any) – The input for the agent, which can be of various types.
timeout (float | None) – Optional time limit in seconds for execution.
trace_params (TraceParams | None) – Optional trace parameters for observability purposes.
- Returns:
The result of the agent execution.
- Return type:
AgentRunOutput[T_Schema]
Example
```python # Input as string result = agent.run(“What is the weather in London?”)
# Input as UserMessage object from agentle.generations.models.messages.user_message import UserMessage from agentle.generations.models.message_parts.text import TextPart
message = UserMessage(parts=[TextPart(text=”What is the weather in London?”)]) result = agent.run(message) ```
- async resume_async(resumption_token, approval_data=None)[source]¶
Resume a suspended agent execution.
- Parameters:
- Returns:
AgentRunOutput with the completed or newly suspended execution
- Raises:
ValueError – If the resumption token is invalid or expired
- Return type:
AgentRunOutput[T_Schema]
- resume(resumption_token, approval_data=None)[source]¶
Resume a suspended agent execution synchronously.
- async run_async(input, *, trace_params=None)[source]¶
Runs the agent asynchronously with the provided input.
This main method processes user input, interacts with the generation provider, and optionally calls tools until reaching a final response.
The method supports both simple agents (without tools) and agents with tools that can perform iterative calls to solve complex tasks.
- Parameters:
input (AgentInput | Any) – The input for the agent, which can be of various types.
trace_params (TraceParams | None) – Optional trace parameters for observability purposes.
- Returns:
- The result of the agent execution, possibly
with a structured response according to the defined schema.
- Return type:
AgentRunOutput[T_Schema]
- Raises:
MaxToolCallsExceededError – If the maximum number of tool calls is exceeded.
Example
```python # Asynchronous use result = await agent.run_async(“What’s the weather like in London?”)
# Processing the response response_text = result.artifacts[0].parts[0].text print(response_text)
# With structured response schema if result.parsed:
location = result.parsed.location weather = result.parsed.weather
- clone(*, new_name=None, new_instructions=None, new_tools=None, new_config=None, new_model=None, new_version=None, new_documentation_url=None, new_capabilities=None, new_authentication=None, new_default_input_modes=None, new_default_output_modes=None, new_skills=None, new_mcp_servers=None, new_generation_provider=None, new_url=None, new_suspension_manager=None, new_document_cache_store=None)[source]¶
Creates a clone of the current agent with optionally modified attributes.
This method facilitates creating variations of an agent without modifying the original. Unspecified parameters will retain the values from the original agent.
- Parameters:
new_name (str | None) – New name for the agent.
new_instructions (str | None) – New instructions for the agent.
new_tools (Sequence[Tool | Callable[..., object]] | None) – New tools for the agent.
new_config (AgentConfig | AgentConfigDict | None) – New configuration for the agent.
new_model (str | None) – New model for the agent.
new_version (str | None) – New version for the agent.
new_documentation_url (str | None) – New documentation URL for the agent.
new_capabilities (Capabilities | None) – New capabilities for the agent.
new_authentication (Authentication | None) – New authentication for the agent.
new_default_input_modes (Sequence[str] | None) – New default input modes for the agent.
new_default_output_modes (Sequence[str] | None) – New default output modes for the agent.
new_skills (Sequence[AgentSkill] | None) – New skills for the agent.
new_mcp_servers (Sequence[MCPServerProtocol] | None) – New MCP servers for the agent.
new_generation_provider (GenerationProvider | None) – New generation provider for the agent.
new_url (str | None) – New URL for the agent.
new_suspension_manager (SuspensionManager | None) – New suspension manager for the agent.
new_document_cache_store (DocumentCacheStore | None) – New cache store for the agent.
- Returns:
A new agent with the specified attributes modified.
- Return type:
Agent[T_Schema]
Example
```python # Create a variation of the agent with different instructions weather_agent_fr = weather_agent.clone(
new_name=”French Weather Agent”, new_instructions=”You are a weather agent that can answer questions about the weather in French.”
)¶
Agent Pipeline¶
Agent Pipeline Module for the Agentle Framework
This module provides functionality for creating sequential pipelines of AI agents, where the output of one agent becomes the input to the next. Pipelines offer a deterministic way to decompose complex tasks into a series of simpler steps handled by specialized agents.
Unlike the AgentTeam which uses an orchestrator to dynamically select agents, AgentPipeline follows a fixed sequence of predefined agents, providing a more predictable execution flow. This is particularly useful for workflows where:
The sequence of operations is known in advance
Each step builds upon the results of the previous step
A clear division of responsibility between agents is needed
Example: ```python from agentle.agents.agent import Agent from agentle.agents.pipelines.agent_pipeline import AgentPipeline from agentle.generations.providers.google.google_genai_generation_provider import GoogleGenaiGenerationProvider
# Create specialized agents research_agent = Agent(
name=”Research Agent”, instructions=”You are a research agent. Your task is to gather information on a topic.”, model=”gemini-2.0-flash”, generation_provider=GoogleGenaiGenerationProvider(),
)
- analysis_agent = Agent(
name=”Analysis Agent”, instructions=”You are an analysis agent. Your task is to analyze information and identify patterns.”, model=”gemini-2.0-flash”, generation_provider=GoogleGenaiGenerationProvider(),
)
- summary_agent = Agent(
name=”Summary Agent”, instructions=”You are a summary agent. Your task is to create concise summaries.”, model=”gemini-2.0-flash”, generation_provider=GoogleGenaiGenerationProvider(),
)
# Create a pipeline of agents pipeline = AgentPipeline(agents=[research_agent, analysis_agent, summary_agent])
# Run the pipeline result = pipeline.run(“Tell me about renewable energy technologies”) print(result.generation.text) ```
- class AgentPipeline(*, agents, debug_mode=False)[source]¶
Bases:
BaseModelA sequential pipeline of AI agents where the output of one agent becomes the input to the next.
AgentPipeline provides a way to chain multiple agents together in a fixed sequence, allowing complex tasks to be broken down into manageable steps. Each agent in the pipeline specializes in a specific subtask, and the agents work together by passing information from one to the next.
Key features: - Sequential execution: Agents run in the exact order specified - Automatic input/output handling: Output from one agent becomes input to the next - Deterministic workflow: The same pipeline with the same initial input produces consistent results - Debug capabilities: Optional logging of intermediate steps for troubleshooting
- agents¶
A sequence of Agent instances that form the pipeline.
- Type:
Examples
```python # Create a simple translation pipeline english_to_french = Agent(
name=”English to French”, instructions=”Translate English text to French.”, model=”gemini-2.0-flash”
)
- french_to_spanish = Agent(
name=”French to Spanish”, instructions=”Translate French text to Spanish.”, model=”gemini-2.0-flash”
)
# Chain the agents into a pipeline translation_pipeline = AgentPipeline(
agents=[english_to_french, french_to_spanish]
)
# Run the pipeline result = translation_pipeline.run(“Hello, how are you?”) # Result will be “Hola, ¿cómo estás?” after passing through both agents ```
Creating a complex analysis pipeline: ```python # Data extraction agent extractor = Agent(name=”Extractor”, instructions=”Extract key data points…”)
# Analysis agent analyzer = Agent(name=”Analyzer”, instructions=”Analyze the following data…”)
# Recommendation agent recommender = Agent(name=”Recommender”, instructions=”Based on the analysis…”)
# Create analysis pipeline analysis_pipeline = AgentPipeline(
agents=[extractor, analyzer, recommender], debug_mode=True # Enable logging for debugging
)
# Run pipeline with initial data result = analysis_pipeline.run(“Raw data: …”) ```
- async resume_async(resumption_token, approval_data=None)[source]¶
Resume a suspended pipeline execution.
This method resumes a pipeline that was suspended due to a tool requiring human approval. It retrieves the pipeline state and continues execution from where it left off.
- Parameters:
- Returns:
AgentRunOutput with the completed or newly suspended execution
- Raises:
ValueError – If the resumption token is invalid or not for a pipeline
- Return type:
AgentRunOutput[Any]
- resume(resumption_token, approval_data=None)[source]¶
Resume a suspended pipeline execution synchronously.
- run(input)[source]¶
Run the agent pipeline synchronously with the provided input.
This method is a synchronous wrapper around run_async that allows the pipeline to be used in synchronous contexts.
- Parameters:
input (AgentInput | Any) – The initial input to the first agent in the pipeline. Can be a string, UserMessage, Context, or any supported AgentInput type.
- Returns:
The output from the final agent in the pipeline.
- Return type:
AgentRunOutput
Example
```python pipeline = AgentPipeline(agents=[agent1, agent2, agent3])
# Simple string input result = pipeline.run(“Analyze this text”)
# Using a more complex input type from agentle.generations.models.messages.user_message import UserMessage from agentle.generations.models.message_parts.text import TextPart
message = UserMessage(parts=[TextPart(text=”Analyze this data”)]) result = pipeline.run(message) ```
- async run_async(input)[source]¶
Executes a pipeline of agents in sequence asynchronously.
This method processes the agents in the pipeline sequentially, where the output of one agent becomes the input to the next agent. The method returns the output of the final agent in the pipeline, or the last agent that successfully produced output.
- Parameters:
input (AgentInput | Any) – The initial input to the first agent in the pipeline
- Returns:
The result from the final agent in the pipeline
- Return type:
AgentRunOutput
- Raises:
ValueError – If the pipeline contains no agents
RuntimeError – If the pipeline execution fails to produce any output
Example
```python pipeline = AgentPipeline(agents=[agent1, agent2, agent3])
# Using with async/await import asyncio
- async def process_data():
result = await pipeline.run_async(“Process this data”) return result.generation.text
final_result = asyncio.run(process_data()) ```
Notes
If an agent in the middle of the pipeline produces no text output, the pipeline will terminate early and return the output of that agent.
The debug_mode attribute controls whether detailed logging is performed during pipeline execution.
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
Agent Team¶
Agent Team Module for the Agentle Framework
This module provides functionality for creating dynamic teams of AI agents managed by an orchestrator. AgentTeam implements a flexible approach to multi-agent systems where agents are selected at runtime based on the specific requirements of each task and subtask.
Unlike the AgentPipeline which follows a fixed sequence of agents, AgentTeam uses an orchestrator agent to dynamically analyze tasks and select the most appropriate agent to handle each step. This provides greater flexibility and adaptability, particularly for complex workflows where:
The optimal sequence of agents depends on the specific task
Different agents may be needed depending on the content of previous responses
The same agent may need to be invoked multiple times for different aspects of a task
The task completion criteria can only be determined at runtime
Example: ```python from agentle.agents.agent import Agent from agentle.agents.agent_team import AgentTeam from agentle.agents.agent_config import AgentConfig from agentle.generations.providers.google.google_genai_generation_provider import GoogleGenaiGenerationProvider
# Create provider for all agents provider = GoogleGenaiGenerationProvider()
# Create specialized agents research_agent = Agent(
name=”Research Agent”, description=”Specialized in finding information and data on various topics”, generation_provider=provider, model=”gemini-2.0-flash”, instructions=”You are a research agent focused on gathering accurate information.”,
)
- coding_agent = Agent(
name=”Coding Agent”, description=”Specialized in writing and debugging code in various languages”, generation_provider=provider, model=”gemini-2.0-flash”, instructions=”You are a coding expert that writes clean, efficient code.”,
)
- writing_agent = Agent(
name=”Writing Agent”, description=”Specialized in creating clear and engaging written content”, generation_provider=provider, model=”gemini-2.0-flash”, instructions=”You are a writing expert that creates compelling content.”,
)
# Create a team with these agents and an orchestrator team = AgentTeam(
agents=[research_agent, coding_agent, writing_agent], orchestrator_provider=provider, orchestrator_model=”gemini-2.0-flash”, team_config=AgentConfig(maxIterations=10)
)
# Run the team with a task result = team.run(“Research the latest advancements in quantum computing and summarize them.”) print(result.generation.text) ```
- class AgentTeam(*, agents, orchestrator_provider=<factory>, orchestrator_model, team_config=<factory>)[source]¶
Bases:
BaseModelA dynamic team of AI agents managed by an orchestrator agent.
AgentTeam implements an approach to multi-agent systems where agents are selected dynamically at runtime by an orchestrator. The orchestrator analyzes each task or subtask and determines which specialized agent is best suited to handle it.
This provides greater flexibility than sequential pipelines, as the exact sequence of agent invocations can be determined based on the specifics of the task and the content of intermediate responses.
Key features: - Dynamic agent selection: The orchestrator chooses the most appropriate agent for each step - Adaptable workflows: The sequence of agents is determined at runtime based on the task - Conversation history: Context is maintained throughout the task execution - Configurable execution limits: Prevents infinite loops with clear termination criteria
- Parameters:
- agents¶
A sequence of specialized Agent instances available to the team.
- Type:
- orchestrator_provider¶
The generation provider used by the orchestrator agent.
- Type:
agentle.generations.providers.base.generation_provider.GenerationProvider
- team_config¶
Configuration options for the team, including iteration limits.
- Type:
agentle.agents.agent_config.AgentConfig
Examples
```python # Create a basic team with several specialized agents provider = GoogleGenaiGenerationProvider()
- math_agent = Agent(
name=”Math Agent”, description=”Expert in solving mathematical problems”, generation_provider=provider, model=”gemini-2.0-flash”, instructions=”You are a mathematics expert.”
)
- language_agent = Agent(
name=”Language Agent”, description=”Expert in language translation and linguistics”, generation_provider=provider, model=”gemini-2.0-flash”, instructions=”You are a language and translation expert.”
)
- team = AgentTeam(
agents=[math_agent, language_agent], orchestrator_provider=provider, orchestrator_model=”gemini-2.0-flash”
)
# Run the team on a task result = team.run(“Translate the phrase ‘The square root of 144 is 12’ into French”) ```
Creating a team with custom configuration: ```python # Create a team with custom configuration custom_team = AgentTeam(
agents=[agent1, agent2, agent3], orchestrator_provider=provider, orchestrator_model=”gemini-2.0-flash”, team_config=AgentConfig(
maxIterations=15, # Allow more iterations than default maxToolCalls=20 # Allow more tool calls if agents use tools
)
)
# Run the team with a complex multi-step task result = custom_team.run(“Research, analyze, and summarize recent developments in AI safety”) ```
- orchestrator_provider: GenerationProvider¶
- team_config: AgentConfig¶
- async resume_async(resumption_token, approval_data=None)[source]¶
Resume a suspended team execution.
This method resumes a team that was suspended due to a tool requiring human approval. It retrieves the team state and continues execution from where it left off.
- Parameters:
- Returns:
AgentRunOutput with the completed or newly suspended execution
- Raises:
ValueError – If the resumption token is invalid or not for a team
- Return type:
AgentRunOutput[Any]
- resume(resumption_token, approval_data=None)[source]¶
Resume a suspended team execution synchronously.
- run(input)[source]¶
Run the agent team synchronously with the provided input.
This method is a synchronous wrapper around run_async that allows the team to be used in synchronous contexts.
- Parameters:
input (AgentInput | Any) – The initial input to the team. Can be a string, UserMessage, Context, or any supported AgentInput type.
- Returns:
The final output from the team after task completion.
- Return type:
AgentRunOutput
Example
agents=[research_agent, writing_agent, code_agent], orchestrator_provider=provider, orchestrator_model=”gemini-2.0-flash”
)
# Simple string input result = team.run(“Create a Python function to calculate the Fibonacci sequence”)
- async run_async(input)[source]¶
Dynamically executes a team of agents guided by an orchestrator.
This method creates an orchestrator agent that analyzes each task step and selects the most appropriate agent to handle it. The process continues iteratively until the orchestrator determines the task is complete or the maximum number of iterations is reached.
Process flow: 1. Orchestrator analyzes the current input/task 2. Orchestrator selects the most appropriate agent and indicates if task is complete 3. If task is complete, return the latest result 4. Otherwise, the selected agent processes the input 5. The agent’s output becomes the next input 6. The process repeats until completion or max iterations
- Parameters:
input (AgentInput | Any) – The initial input to the team
- Returns:
The final result when the task is complete
- Return type:
AgentRunOutput
- Raises:
ValueError – If the team contains no agents
Example
```python # Using the team with async/await import asyncio
- async def process_complex_task():
- team = AgentTeam(
agents=[agent1, agent2, agent3], orchestrator_provider=provider, orchestrator_model=”gemini-2.0-flash”
)
result = await team.run_async(“Perform this complex multi-step task”) return result.generation.text
final_result = asyncio.run(process_complex_task()) ```
Notes
The team_config.maxIterations parameter controls the maximum number of agent invocations to prevent infinite loops.
Conversation history is maintained to provide context to the orchestrator.
The orchestrator has complete information about all available agents.
- __add__(other)[source]¶
Combine this AgentTeam with another Agent or AgentTeam.
This operator allows for easy composition of teams by combining their agents. When adding an Agent, it’s incorporated into the current team’s agent list. When adding another AgentTeam, all its agents are added to the current team.
- Parameters:
other (Agent[Any] | AgentTeam) – Another Agent or AgentTeam to combine with this team
- Returns:
A new AgentTeam containing all agents from both sources
- Return type:
Example
```python # Create initial team basic_team = AgentTeam(
agents=[agent1, agent2], orchestrator_provider=provider, orchestrator_model=”gemini-2.0-flash”
)
# Add another agent expanded_team = basic_team + specialized_agent
# Add another team other_team = AgentTeam(agents=[agent3, agent4], …) combined_team = expanded_team + other_team
# The combined team now contains all agents from both teams ```
- model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].