diff --git a/.cursorrules b/.cursorrules index 167509b5ff..708536d98a 100644 --- a/.cursorrules +++ b/.cursorrules @@ -6,6 +6,10 @@ Core Rules - PostgreSQL in production, SQLite for dev only - Start with single agent, scale up only when needed +Documentation: +- Don't use f-strings for print lines where there are no variables to format. +- Don't use emojis in examples and print lines + Basic Agent (start here): ```python from agno.agent import Agent diff --git a/cookbook/agent_os/background_tasks/README.md b/cookbook/agent_os/background_tasks/README.md new file mode 100644 index 0000000000..962151d338 --- /dev/null +++ b/cookbook/agent_os/background_tasks/README.md @@ -0,0 +1,107 @@ +# Background Tasks in AgentOS + +This directory contains examples demonstrating how to run hooks as background tasks in AgentOS. Background hooks execute after the API response is sent to the user, making them non-blocking. + +## Why Use Background Hooks? + +Background hooks are useful when you need to perform operations that: +- Don't affect the response to the user +- Can tolerate eventual execution +- Would otherwise add latency to the API response + +Common use cases: +- Logging and analytics +- Sending notifications +- Quality evaluation and monitoring +- Storing metrics to a database +- Triggering downstream workflows + +## Two Approaches + +### 1. Global Setting (`run_hooks_in_background`) + +Enable background execution for all hooks at the AgentOS level: + +```python +agent_os = AgentOS( + agents=[agent], + run_hooks_in_background=True, # All hooks run in background +) +``` + +See: `background_hooks_example.py` + +### 2. Per-Hook Decorator (`@hook`) + +Control background execution per hook using the decorator: + +```python +from agno.hooks.decorator import hook + +@hook(run_in_background=True) +async def my_background_hook(run_output, agent): + # This runs in background regardless of global setting + await send_notification(run_output) + +async def my_blocking_hook(run_output, agent): + # This runs normally (blocking) + validate_output(run_output) +``` + +See: `background_hooks_decorator.py` + +## Examples + +| File | Description | +|------|-------------| +| `background_hooks_example.py` | Basic example using global `run_hooks_in_background=True` | +| `background_hooks_decorator.py` | Per-hook control using `@hook(run_in_background=True)` | +| `background_hooks_team.py` | Background hooks with a Team | +| `background_hooks_workflow.py` | Background hooks with a Workflow | +| `background_output_evaluation.py` | Agent-as-judge pattern for quality monitoring | + +## Running the Examples + +Start any example server: + +```bash +python background_hooks_example.py +``` + +Test with a request: + +```bash +curl -X POST http://localhost:7777/agents/{agent-id}/runs \ + -F "message=Hello!" \ + -F "stream=false" +``` + +The response returns immediately while background hooks continue executing. + +## Background Output Evaluation + +The `background_output_evaluation.py` example demonstrates using an evaluator agent to assess response quality without blocking: + +```python +@hook(run_in_background=True) +async def evaluate_output_quality(run_output: RunOutput, agent: Agent) -> None: + result = await evaluator_agent.arun(input=f"Evaluate: {run_output.content}") + # Log results for monitoring +``` + +This pattern is useful for: +- Production quality monitoring +- A/B testing response quality +- Compliance auditing +- Building evaluation datasets + +## Important Notes + +1. **Pre-hooks in background mode cannot modify `run_input`** - modifications won't affect the agent run since it may have already started. + +2. **Background hooks cannot block responses** - use blocking hooks if you need to validate/reject outputs. + +3. **Error handling** - background hook errors are logged but don't affect the API response. + +4. **Agent reuse** - create evaluator agents outside hooks to avoid recreation overhead on every request. + diff --git a/cookbook/agent_os/background_tasks/background_evals_example.py b/cookbook/agent_os/background_tasks/background_evals_example.py new file mode 100644 index 0000000000..2518975e17 --- /dev/null +++ b/cookbook/agent_os/background_tasks/background_evals_example.py @@ -0,0 +1,81 @@ +""" +Example: Per-Hook Background Control in AgentOS + +This example demonstrates fine-grained control over which hooks run in background: +- Set eval.run_in_background = True for eval instances +""" + +from agno.agent import Agent +from agno.db.sqlite import AsyncSqliteDb +from agno.eval.performance import PerformanceEval +from agno.eval.reliability import ReliabilityEval +from agno.models.openai import OpenAIChat +from agno.os import AgentOS +from agno.tools.calculator import CalculatorTools + +# Setup database +db = AsyncSqliteDb(db_file="tmp/evals.db") + + +# Function to benchmark +async def count_to_100(): + import asyncio + + await asyncio.sleep(2) + total = 0 + for i in range(1, 101): + total += i + return total + + +# PerformanceEval - runs in background +performance_eval = PerformanceEval( + func=count_to_100, + num_iterations=5, + warmup_runs=2, + db=db, + print_results=False, + print_summary=True, # Just show summary + telemetry=False, +) +performance_eval.run_in_background = True + +# ReliabilityEval - runs synchronously (default) +reliability_eval = ReliabilityEval( + expected_tool_calls=["add", "multiply", "divide"], + db=db, + print_results=True, + telemetry=False, +) +# reliability_eval.run_in_background = True + +agent = Agent( + id="math-agent", + name="MathAgent", + model=OpenAIChat(id="gpt-4o-mini"), + instructions="You are a helpful math assistant. Always use the Calculator tools.", + tools=[CalculatorTools()], + db=db, + post_hooks=[ + reliability_eval, # run_in_background=False - runs first, blocks + performance_eval, # run_in_background=True - runs after response + ], + markdown=True, + telemetry=False, +) + +# Create AgentOS +agent_os = AgentOS(agents=[agent]) +app = agent_os.get_app() + +# Flow: +# 1. Agent processes request +# 2. Sync hooks run (reliability_eval) +# 3. Response sent to user +# 4. Background hooks run (performance_eval) + +# curl -X POST http://localhost:7777/agents/math-agent/runs \ +# -F "message=What is 2+2?" -F "stream=false" + +if __name__ == "__main__": + agent_os.serve(app="background_evals_example:app", port=7777, reload=True) diff --git a/cookbook/agent_os/background_tasks/background_hooks_decorator.py b/cookbook/agent_os/background_tasks/background_hooks_decorator.py new file mode 100644 index 0000000000..d3caf87c9a --- /dev/null +++ b/cookbook/agent_os/background_tasks/background_hooks_decorator.py @@ -0,0 +1,82 @@ +""" +Example: Using Background Post-Hooks in AgentOS + +This example demonstrates how to run post-hooks as FastAPI background tasks, +making them completely non-blocking. +""" + +import asyncio + +from agno.agent import Agent +from agno.db.sqlite import AsyncSqliteDb +from agno.hooks.decorator import hook +from agno.models.openai import OpenAIChat +from agno.os import AgentOS +from agno.run.agent import RunInput + + +@hook(run_in_background=True) +def log_request(run_input: RunInput, agent): + """ + This pre-hook will run in the background before the agent processes the request. + Note: Pre-hooks in background mode cannot modify run_input. + """ + print(f"[Background Pre-Hook] Request received for agent: {agent.name}") + print(f"[Background Pre-Hook] Input: {run_input.input_content}") + + +async def log_analytics(run_output, agent, session): + """ + Post hook for logging analytics + """ + print(f"[Post-Hook] Logging analytics for run: {run_output.run_id}") + print(f"[Post-Hook] Agent: {agent.name}") + print(f"[Post-Hook] Session: {session.session_id}") + print("[Post-Hook] Analytics logged successfully!") + + +@hook(run_in_background=True) +async def send_notification(run_output, agent): + """ + Post hook for sending notifications + """ + print(f"[Post-Hook] Sending notification for agent: {agent.name}") + await asyncio.sleep(3) + print("[Post-Hook] Notification sent!") + + +# Create an agent with background post-hooks enabled +agent = Agent( + id="background-task-agent", + name="BackgroundTaskAgent", + model=OpenAIChat(id="gpt-5-mini"), + instructions="You are a helpful assistant", + db=AsyncSqliteDb(db_file="tmp/agent.db"), + # Define hooks + pre_hooks=[log_request], + post_hooks=[log_analytics, send_notification], + markdown=True, +) + +# Create AgentOS +agent_os = AgentOS( + agents=[agent], +) + +# Get the FastAPI app +app = agent_os.get_app() + + +# When you make a request to POST /agents/{agent_id}/runs: +# 1. The agent will process the request +# 2. The response will be sent immediately to the user, with log_analytics executing with the normal execution flow +# 3. The pre-hooks (log_request) and post-hooks (send_notification) will run in the background without blocking the API response +# 4. The user doesn't have to wait for these tasks to complete + +# Example request: +# curl -X POST http://localhost:8000/agents/background-task-agent/runs \ +# -F "message=Hello, how are you?" \ +# -F "stream=false" + +if __name__ == "__main__": + agent_os.serve(app="background_hooks_decorator:app", port=7777, reload=True) diff --git a/cookbook/agent_os/background_tasks/background_hooks_example.py b/cookbook/agent_os/background_tasks/background_hooks_example.py new file mode 100644 index 0000000000..db36d3a93a --- /dev/null +++ b/cookbook/agent_os/background_tasks/background_hooks_example.py @@ -0,0 +1,92 @@ +""" +Example: Using Background Post-Hooks in AgentOS + +This example demonstrates how to run post-hooks as FastAPI background tasks, +making them completely non-blocking. +""" + +import asyncio + +from agno.agent import Agent +from agno.db.sqlite import AsyncSqliteDb +from agno.models.openai import OpenAIChat +from agno.os import AgentOS +from agno.run.agent import RunInput + + +# Pre-hook for logging request +def log_request(run_input: RunInput, agent): + """ + This pre-hook will run in the background before the agent processes the request. + Note: Pre-hooks in background mode cannot modify run_input. + """ + print(f"[Background Pre-Hook] Request received for agent: {agent.name}") + print(f"[Background Pre-Hook] Input: {run_input.input_content}") + + +# Post-hook for logging analytics +async def log_analytics(run_output, agent, session): + """ + This post-hook will run in the background after the response is sent to the user. + It won't block the API response. + """ + print(f"[Background Post-Hook] Logging analytics for run: {run_output.run_id}") + print(f"[Background Post-Hook] Agent: {agent.name}") + print(f"[Background Post-Hook] Session: {session.session_id}") + + # Simulate a delay of 2 seconds + await asyncio.sleep(2) + print("[Background Post-Hook] Analytics logged successfully!") + + +# Another post-hook for sending notifications +async def send_notification(run_output, agent): + """ + Another background task that sends notifications without blocking the response. + """ + print(f"[Background Post-Hook] Sending notification for agent: {agent.name}") + # Simulate a delay of 3 seconds + await asyncio.sleep(3) + print("[Background Post-Hook] Notification sent!") + + +# Create an agent with background post-hooks enabled +agent = Agent( + id="background-task-agent", + name="BackgroundTaskAgent", + model=OpenAIChat(id="gpt-5-mini"), + instructions="You are a helpful assistant", + db=AsyncSqliteDb(db_file="tmp/agent.db"), + # Define hooks + pre_hooks=[log_request], + post_hooks=[log_analytics, send_notification], + markdown=True, +) + +# Create AgentOS +agent_os = AgentOS( + agents=[agent], + # Enable background mode for hooks (if you disable, the hooks will run in the main thread) + run_hooks_in_background=True, +) + +# Get the FastAPI app +app = agent_os.get_app() + + +# When you make a request to POST /agents/{agent_id}/runs: +# 1. The agent will process the request +# 2. The response will be sent immediately to the user +# 3. The pre-hooks (log_request) and post-hooks (log_analytics, send_notification) will run in the background +# 4. The user doesn't have to wait for these tasks to complete + +# Example request: +# curl -X POST http://localhost:8000/agents/background-task-agent/runs \ +# -F "message=Hello, how are you?" \ +# -F "stream=false" + +# The response will be returned immediately, while log_request, log_analytics and send_notification +# continue to run in the background without blocking the API response. + +if __name__ == "__main__": + agent_os.serve(app="background_hooks_example:app", port=7777, reload=True) diff --git a/cookbook/agent_os/background_tasks/background_hooks_team.py b/cookbook/agent_os/background_tasks/background_hooks_team.py new file mode 100644 index 0000000000..20374e3a80 --- /dev/null +++ b/cookbook/agent_os/background_tasks/background_hooks_team.py @@ -0,0 +1,72 @@ +""" +Example: Background Hooks with Teams in AgentOS + +This example demonstrates how to use background hooks with a Team. +Background hooks execute after the API response is sent, making them non-blocking. +""" + +import asyncio + +from agno.agent import Agent +from agno.db.sqlite import AsyncSqliteDb +from agno.hooks.decorator import hook +from agno.models.openai import OpenAIChat +from agno.os import AgentOS +from agno.run.team import TeamRunOutput +from agno.team import Team + + +@hook(run_in_background=True) +async def log_team_result(run_output: TeamRunOutput, team: Team) -> None: + """ + Background post-hook that logs team execution results. + Runs after the response is sent to the user. + """ + print(f"[Background Hook] Team '{team.name}' completed run: {run_output.run_id}") + print(f"[Background Hook] Content length: {len(str(run_output.content))} chars") + + # Simulate async work (e.g., storing metrics) + await asyncio.sleep(2) + print("[Background Hook] Team metrics logged successfully!") + + +# Create team members +researcher = Agent( + name="Researcher", + model=OpenAIChat(id="gpt-4o-mini"), + instructions="You research topics and provide factual information.", +) + +writer = Agent( + name="Writer", + model=OpenAIChat(id="gpt-4o-mini"), + instructions="You write clear, engaging content based on research.", +) + +# Create the team with background hooks +content_team = Team( + id="content-team", + name="ContentTeam", + model=OpenAIChat(id="gpt-4o-mini"), + members=[researcher, writer], + instructions="Coordinate between researcher and writer to create content.", + db=AsyncSqliteDb(db_file="tmp/team.db"), + post_hooks=[log_team_result], + markdown=True, +) + +# Create AgentOS with background hooks enabled +agent_os = AgentOS( + teams=[content_team], + run_hooks_in_background=True, +) + +app = agent_os.get_app() + +# Example request: +# curl -X POST http://localhost:7777/teams/content-team/runs \ +# -F "message=Write a short paragraph about Python" \ +# -F "stream=false" + +if __name__ == "__main__": + agent_os.serve(app="background_hooks_team:app", port=7777, reload=True) diff --git a/cookbook/agent_os/background_tasks/background_hooks_workflow.py b/cookbook/agent_os/background_tasks/background_hooks_workflow.py new file mode 100644 index 0000000000..86137c3f79 --- /dev/null +++ b/cookbook/agent_os/background_tasks/background_hooks_workflow.py @@ -0,0 +1,68 @@ +""" +Example: Background Hooks with Workflows in AgentOS + +This example demonstrates how to use background hooks with a Workflow. +Background hooks execute after the API response is sent, making them non-blocking. +""" + +import asyncio + +from agno.agent import Agent +from agno.db.sqlite import AsyncSqliteDb +from agno.models.openai import OpenAIChat +from agno.os import AgentOS +from agno.run.agent import RunOutput +from agno.workflow import Workflow + + +async def log_step_completion(run_output: RunOutput, agent: Agent) -> None: + """ + Background post-hook on the agent that runs after each step completes. + """ + print(f"[Background Hook] Agent '{agent.name}' completed step") + print(f"[Background Hook] Run ID: {run_output.run_id}") + + # Simulate async work + await asyncio.sleep(1) + print(f"[Background Hook] Logged metrics for {agent.name}") + + +# Create agents for the workflow steps +analyzer = Agent( + name="Analyzer", + model=OpenAIChat(id="gpt-4o-mini"), + instructions="Analyze the input and identify key points.", + post_hooks=[log_step_completion], +) + +summarizer = Agent( + name="Summarizer", + model=OpenAIChat(id="gpt-4o-mini"), + instructions="Summarize the analysis into a brief response.", + post_hooks=[log_step_completion], +) + +# Create the workflow +analysis_workflow = Workflow( + id="analysis-workflow", + name="AnalysisWorkflow", + description="Analyzes input and provides a summary", + steps=[analyzer, summarizer], + db=AsyncSqliteDb(db_file="tmp/workflow.db"), +) + +# Create AgentOS with background hooks enabled +agent_os = AgentOS( + workflows=[analysis_workflow], + run_hooks_in_background=True, +) + +app = agent_os.get_app() + +# Example request: +# curl -X POST http://localhost:7777/workflows/analysis-workflow/runs \ +# -F "message=Explain the benefits of exercise" \ +# -F "stream=false" + +if __name__ == "__main__": + agent_os.serve(app="background_hooks_workflow:app", port=7777, reload=True) diff --git a/cookbook/agent_os/background_tasks/background_output_evaluation.py b/cookbook/agent_os/background_tasks/background_output_evaluation.py new file mode 100644 index 0000000000..53789eb043 --- /dev/null +++ b/cookbook/agent_os/background_tasks/background_output_evaluation.py @@ -0,0 +1,192 @@ +""" +Example: Background Output Evaluation with Agent-as-Judge + +This example demonstrates how to use a validator agent to evaluate the main agent's +output as a background task. Unlike blocking validation, background evaluation: +- Does NOT block the response to the user +- Logs evaluation results for monitoring and analytics +- Can trigger alerts or store metrics without affecting latency + +Use cases: +- Quality monitoring in production +- A/B testing response quality +- Compliance auditing +- Building evaluation datasets +""" + +import asyncio +from datetime import datetime + +from agno.agent import Agent +from agno.db.sqlite import AsyncSqliteDb +from agno.hooks.decorator import hook +from agno.models.openai import OpenAIChat +from agno.os import AgentOS +from agno.run.agent import RunOutput +from pydantic import BaseModel + + +class EvaluationResult(BaseModel): + """Structured output for the evaluator agent.""" + + is_helpful: bool + is_accurate: bool + is_well_structured: bool + quality_score: float # 0.0 - 1.0 + strengths: list[str] + areas_for_improvement: list[str] + summary: str + + +# Create the evaluator agent once (not in the hook for performance) +evaluator_agent = Agent( + name="OutputEvaluator", + model=OpenAIChat(id="gpt-4o-mini"), + instructions=[ + "You are an expert at evaluating AI assistant responses.", + "Analyze responses for:", + "1. HELPFULNESS: Does it address the user's question?", + "2. ACCURACY: Is the information correct and reliable?", + "3. STRUCTURE: Is it well-organized and easy to understand?", + "", + "Provide a quality_score from 0.0 to 1.0 where:", + "- 0.0-0.3: Poor quality, major issues", + "- 0.4-0.6: Acceptable, some improvements needed", + "- 0.7-0.8: Good quality, minor issues", + "- 0.9-1.0: Excellent quality", + "", + "Be fair and balanced in your evaluation.", + ], + output_schema=EvaluationResult, +) + + +@hook(run_in_background=True) +async def evaluate_output_quality(run_output: RunOutput, agent: Agent) -> None: + """ + Background post-hook that evaluates the agent's response quality. + + This runs after the response is sent to the user, so it doesn't add latency. + Results are logged for monitoring purposes. + """ + # Skip if no content to evaluate + if not run_output.content or len(str(run_output.content).strip()) < 10: + print("[Evaluator] Skipping evaluation - response too short") + return + + print(f"[Evaluator] Starting background evaluation for run: {run_output.run_id}") + + # Run the evaluation + evaluation_prompt = f""" + Evaluate this AI assistant response: + + User Query: {run_output.input_content if hasattr(run_output, "input_content") else "Unknown"} + + Assistant Response: + {run_output.content} + """ + + result = await evaluator_agent.arun(input=evaluation_prompt) + evaluation: EvaluationResult = result.content + + # Log the evaluation results + timestamp = datetime.now().isoformat() + print("\n" + "=" * 60) + print(f"[Evaluator] Evaluation Complete - {timestamp}") + print("=" * 60) + print(f"Run ID: {run_output.run_id}") + print(f"Agent: {agent.name}") + print(f"\nQuality Score: {evaluation.quality_score:.2f}/1.00") + print(f"Helpful: {evaluation.is_helpful}") + print(f"Accurate: {evaluation.is_accurate}") + print(f"Well-Structured: {evaluation.is_well_structured}") + + if evaluation.strengths: + print("\nStrengths:") + for strength in evaluation.strengths: + print(f" - {strength}") + + if evaluation.areas_for_improvement: + print("\nAreas for Improvement:") + for area in evaluation.areas_for_improvement: + print(f" - {area}") + + print(f"\nSummary: {evaluation.summary}") + print("=" * 60 + "\n") + + # In production, you could: + # - Store in database for analytics + # - Send alerts if quality_score < threshold + # - Log to observability platform + # - Build evaluation datasets + + +@hook(run_in_background=True) +async def check_response_safety(run_output: RunOutput, agent: Agent) -> None: + """ + Background post-hook that checks response safety. + + Runs concurrently with other background hooks. + """ + print(f"[Safety Check] Analyzing response safety for run: {run_output.run_id}") + + # Simulate safety check processing + await asyncio.sleep(1) + + content = str(run_output.content).lower() + + # Simple safety checks (in production, use a more sophisticated approach) + safety_flags = [] + if any(word in content for word in ["password", "credential", "secret"]): + safety_flags.append("Contains potentially sensitive terms") + + if len(safety_flags) > 0: + print(f"[Safety Check] Flags found: {safety_flags}") + else: + print("[Safety Check] No safety concerns detected") + + +# Setup database for agent storage +db = AsyncSqliteDb(db_file="tmp/evaluation.db") + +# Create the main agent with background evaluation hooks +main_agent = Agent( + id="support-agent", + name="CustomerSupportAgent", + model=OpenAIChat(id="gpt-4o-mini"), + instructions=[ + "You are a helpful customer support agent.", + "Provide clear, accurate, and friendly responses.", + "If you don't know something, say so honestly.", + ], + db=db, + post_hooks=[ + evaluate_output_quality, # Runs in background + check_response_safety, # Runs in background + ], + markdown=True, +) + +# Create AgentOS +agent_os = AgentOS(agents=[main_agent]) +app = agent_os.get_app() + + +# Flow: +# 1. User sends request to /agents/support-agent/runs +# 2. Agent processes and generates response +# 3. Response is sent to user immediately +# 4. Background hooks run concurrently: +# - evaluate_output_quality: Evaluator agent scores the response +# - check_response_safety: Safety checks run in parallel +# 5. Evaluation results are logged for monitoring + +# Example requests: +# curl -X POST http://localhost:7777/agents/support-agent/runs \ +# -F "message=How do I reset my password?" -F "stream=false" +# +# curl -X POST http://localhost:7777/agents/support-agent/runs \ +# -F "message=Explain the difference between HTTP and HTTPS" -F "stream=false" + +if __name__ == "__main__": + agent_os.serve(app="background_output_evaluation:app", port=7777, reload=True) diff --git a/libs/agno/agno/agent/agent.py b/libs/agno/agno/agent/agent.py index 490a1f4a4f..bb6c89fa1c 100644 --- a/libs/agno/agno/agent/agent.py +++ b/libs/agno/agno/agent/agent.py @@ -83,8 +83,8 @@ from agno.utils.agent import ( aget_session_state_util, aset_session_name_util, aupdate_session_state_util, - await_for_background_tasks, - await_for_background_tasks_stream, + await_for_open_threads, + await_for_thread_tasks_stream, collect_joint_audios, collect_joint_files, collect_joint_images, @@ -103,8 +103,8 @@ from agno.utils.agent import ( store_media_util, update_session_state_util, validate_media_object_id, - wait_for_background_tasks, - wait_for_background_tasks_stream, + wait_for_open_threads, + wait_for_thread_tasks_stream, ) from agno.utils.common import is_typed_dict, validate_typed_dict from agno.utils.events import ( @@ -131,7 +131,7 @@ from agno.utils.events import ( create_tool_call_started_event, handle_event, ) -from agno.utils.hooks import filter_hook_args, normalize_hooks +from agno.utils.hooks import copy_args_for_background, filter_hook_args, normalize_hooks, should_run_hook_in_background from agno.utils.knowledge import get_agentic_or_user_search_filters from agno.utils.log import ( log_debug, @@ -273,6 +273,8 @@ class Agent: pre_hooks: Optional[List[Union[Callable[..., Any], BaseGuardrail]]] = None # Functions called after output is generated but before the response is returned post_hooks: Optional[List[Union[Callable[..., Any], BaseGuardrail]]] = None + # If True, run hooks as FastAPI background tasks (non-blocking). Set by AgentOS. + _run_hooks_in_background: Optional[bool] = None # --- Agent Reasoning --- # Enable reasoning by working through the problem step by step. @@ -594,7 +596,6 @@ class Agent: self.tool_choice = tool_choice self.tool_hooks = tool_hooks - # Initialize hooks with backward compatibility self.pre_hooks = pre_hooks self.post_hooks = post_hooks @@ -971,6 +972,7 @@ class Agent: add_session_state_to_context: Optional[bool] = None, response_format: Optional[Union[Dict, Type[BaseModel]]] = None, debug_mode: Optional[bool] = None, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> RunOutput: """Run the Agent and return the RunOutput. @@ -1007,6 +1009,7 @@ class Agent: session=session, user_id=user_id, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) # Consume the generator without yielding @@ -1109,9 +1112,7 @@ class Agent: # We should break out of the run function if any(tool_call.is_paused for tool_call in run_response.tools or []): - wait_for_background_tasks( - memory_future=memory_future, cultural_knowledge_future=cultural_knowledge_future - ) + wait_for_open_threads(memory_future=memory_future, cultural_knowledge_future=cultural_knowledge_future) return self._handle_agent_run_paused(run_response=run_response, session=session, user_id=user_id) @@ -1131,6 +1132,7 @@ class Agent: session=session, user_id=user_id, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) deque(post_hook_iterator, maxlen=0) @@ -1139,7 +1141,7 @@ class Agent: raise_if_cancelled(run_response.run_id) # type: ignore # 11. Wait for background memory creation and cultural knowledge creation - wait_for_background_tasks(memory_future=memory_future, cultural_knowledge_future=cultural_knowledge_future) + wait_for_open_threads(memory_future=memory_future, cultural_knowledge_future=cultural_knowledge_future) # 12. Create session summary if self.session_summary_manager is not None: @@ -1192,6 +1194,7 @@ class Agent: stream_events: bool = False, yield_run_output: Optional[bool] = None, debug_mode: Optional[bool] = None, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> Iterator[Union[RunOutputEvent, RunOutput]]: """Run the Agent and yield the RunOutput. @@ -1226,6 +1229,7 @@ class Agent: user_id=user_id, debug_mode=debug_mode, stream_events=stream_events, + background_tasks=background_tasks, **kwargs, ) for event in pre_hook_iterator: @@ -1369,7 +1373,7 @@ class Agent: # We should break out of the run function if any(tool_call.is_paused for tool_call in run_response.tools or []): - yield from wait_for_background_tasks_stream( + yield from wait_for_thread_tasks_stream( memory_future=memory_future, cultural_knowledge_future=cultural_knowledge_future, stream_events=stream_events, @@ -1403,11 +1407,12 @@ class Agent: user_id=user_id, debug_mode=debug_mode, stream_events=stream_events, + background_tasks=background_tasks, **kwargs, ) # 8. Wait for background memory creation and cultural knowledge creation - yield from wait_for_background_tasks_stream( + yield from wait_for_thread_tasks_stream( memory_future=memory_future, cultural_knowledge_future=cultural_knowledge_future, stream_events=stream_events, @@ -1601,6 +1606,12 @@ class Agent: stacklevel=2, ) + background_tasks = kwargs.pop("background_tasks", None) + if background_tasks is not None: + from fastapi import BackgroundTasks + + background_tasks: BackgroundTasks = background_tasks # type: ignore + # Create a run_id for this specific run run_id = str(uuid4()) @@ -1754,6 +1765,7 @@ class Agent: stream_events=stream_events, yield_run_output=yield_run_output, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) return response_iterator @@ -1768,6 +1780,7 @@ class Agent: add_session_state_to_context=add_session_state, response_format=response_format, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) return response @@ -1826,6 +1839,7 @@ class Agent: add_session_state_to_context: Optional[bool] = None, response_format: Optional[Union[Dict, Type[BaseModel]]] = None, debug_mode: Optional[bool] = None, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> RunOutput: """Run the Agent and return the RunOutput. @@ -1888,6 +1902,7 @@ class Agent: session=agent_session, user_id=user_id, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) # Consume the async iterator without yielding @@ -1990,9 +2005,7 @@ class Agent: # We should break out of the run function if any(tool_call.is_paused for tool_call in run_response.tools or []): - await await_for_background_tasks( - memory_task=memory_task, cultural_knowledge_task=cultural_knowledge_task - ) + await await_for_open_threads(memory_task=memory_task, cultural_knowledge_task=cultural_knowledge_task) return await self._ahandle_agent_run_paused( run_response=run_response, session=agent_session, user_id=user_id ) @@ -2013,6 +2026,7 @@ class Agent: session=agent_session, user_id=user_id, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ): pass @@ -2021,7 +2035,7 @@ class Agent: raise_if_cancelled(run_response.run_id) # type: ignore # 14. Wait for background memory creation - await await_for_background_tasks(memory_task=memory_task, cultural_knowledge_task=cultural_knowledge_task) + await await_for_open_threads(memory_task=memory_task, cultural_knowledge_task=cultural_knowledge_task) # 15. Create session summary if self.session_summary_manager is not None: @@ -2100,6 +2114,7 @@ class Agent: stream_events: bool = False, yield_run_output: Optional[bool] = None, debug_mode: Optional[bool] = None, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> AsyncIterator[Union[RunOutputEvent, RunOutput]]: """Run the Agent and yield the RunOutput. @@ -2166,6 +2181,7 @@ class Agent: user_id=user_id, debug_mode=debug_mode, stream_events=stream_events, + background_tasks=background_tasks, **kwargs, ) async for event in pre_hook_iterator: @@ -2308,7 +2324,7 @@ class Agent: # Break out of the run function if a tool call is paused if any(tool_call.is_paused for tool_call in run_response.tools or []): - async for item in await_for_background_tasks_stream( + async for item in await_for_thread_tasks_stream( memory_task=memory_task, cultural_knowledge_task=cultural_knowledge_task, stream_events=stream_events, @@ -2332,12 +2348,13 @@ class Agent: user_id=user_id, debug_mode=debug_mode, stream_events=stream_events, + background_tasks=background_tasks, **kwargs, ): yield event # 11. Wait for background memory creation - async for item in await_for_background_tasks_stream( + async for item in await_for_thread_tasks_stream( memory_task=memory_task, cultural_knowledge_task=cultural_knowledge_task, stream_events=stream_events, @@ -2553,6 +2570,12 @@ class Agent: stacklevel=2, ) + background_tasks = kwargs.pop("background_tasks", None) + if background_tasks is not None: + from fastapi import BackgroundTasks + + background_tasks: BackgroundTasks = background_tasks # type: ignore + # Create a run_id for this specific run run_id = str(uuid4()) @@ -2699,6 +2722,7 @@ class Agent: add_dependencies_to_context=add_dependencies, add_session_state_to_context=add_session_state, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) # type: ignore[assignment] else: @@ -2712,6 +2736,7 @@ class Agent: add_dependencies_to_context=add_dependencies, add_session_state_to_context=add_session_state, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) @@ -2843,6 +2868,12 @@ class Agent: if self._has_async_db(): raise Exception("continue_run() is not supported with an async DB. Please use acontinue_arun() instead.") + background_tasks = kwargs.pop("background_tasks", None) + if background_tasks is not None: + from fastapi import BackgroundTasks + + background_tasks: BackgroundTasks = background_tasks # type: ignore + session_id = run_response.session_id if run_response else session_id run_id: str = run_response.run_id if run_response else run_id # type: ignore @@ -2986,6 +3017,7 @@ class Agent: response_format=response_format, stream_events=stream_events, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) return response_iterator @@ -2999,6 +3031,7 @@ class Agent: session=agent_session, response_format=response_format, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) return response @@ -3049,6 +3082,7 @@ class Agent: user_id: Optional[str] = None, response_format: Optional[Union[Dict, Type[BaseModel]]] = None, debug_mode: Optional[bool] = None, + background_tasks: Optional[Any] = None, **kwargs, ) -> RunOutput: """Continue a previous run. @@ -3113,6 +3147,7 @@ class Agent: session=session, user_id=user_id, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) deque(post_hook_iterator, maxlen=0) @@ -3168,6 +3203,7 @@ class Agent: response_format: Optional[Union[Dict, Type[BaseModel]]] = None, stream_events: bool = False, debug_mode: Optional[bool] = None, + background_tasks: Optional[Any] = None, **kwargs, ) -> Iterator[RunOutputEvent]: """Continue a previous run. @@ -3244,6 +3280,7 @@ class Agent: user_id=user_id, debug_mode=debug_mode, stream_events=stream_events, + background_tasks=background_tasks, **kwargs, ) @@ -3345,6 +3382,7 @@ class Agent: dependencies: Optional[Dict[str, Any]] = None, metadata: Optional[Dict[str, Any]] = None, debug_mode: Optional[bool] = None, + **kwargs: Any, ) -> RunOutput: ... @overload @@ -3364,6 +3402,7 @@ class Agent: dependencies: Optional[Dict[str, Any]] = None, metadata: Optional[Dict[str, Any]] = None, debug_mode: Optional[bool] = None, + **kwargs: Any, ) -> AsyncIterator[Union[RunOutputEvent, RunOutput]]: ... def acontinue_run( # type: ignore @@ -3411,6 +3450,12 @@ class Agent: if run_response is None and (run_id is not None and (session_id is None and self.session_id is None)): raise ValueError("Session ID is required to continue a run from a run_id.") + background_tasks = kwargs.pop("background_tasks", None) + if background_tasks is not None: + from fastapi import BackgroundTasks + + background_tasks: BackgroundTasks = background_tasks # type: ignore + session_id, user_id = self._initialize_session( session_id=session_id, user_id=user_id, @@ -3495,6 +3540,7 @@ class Agent: stream_events=stream_events, yield_run_output=yield_run_output, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) else: @@ -3507,6 +3553,7 @@ class Agent: user_id=user_id, response_format=response_format, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) except ModelProviderError as e: @@ -3556,6 +3603,7 @@ class Agent: user_id: Optional[str] = None, response_format: Optional[Union[Dict, Type[BaseModel]]] = None, debug_mode: Optional[bool] = None, + background_tasks: Optional[Any] = None, **kwargs, ) -> RunOutput: """Continue a previous run. @@ -3698,6 +3746,7 @@ class Agent: session=agent_session, user_id=user_id, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ): pass @@ -3767,6 +3816,7 @@ class Agent: stream_events: bool = False, yield_run_output: bool = False, debug_mode: Optional[bool] = None, + background_tasks: Optional[Any] = None, **kwargs, ) -> AsyncIterator[Union[RunOutputEvent, RunOutput]]: """Continue a previous run. @@ -3952,6 +4002,7 @@ class Agent: user_id=user_id, debug_mode=debug_mode, stream_events=stream_events, + background_tasks=background_tasks, **kwargs, ): yield event @@ -4053,13 +4104,13 @@ class Agent: user_id: Optional[str] = None, debug_mode: Optional[bool] = None, stream_events: bool = False, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> Iterator[RunOutputEvent]: """Execute multiple pre-hook functions in succession.""" if hooks is None: return - - # Prepare all possible arguments once + # Prepare arguments for this hook all_args = { "run_input": run_input, "run_context": run_context, @@ -4071,9 +4122,32 @@ class Agent: "user_id": user_id, "debug_mode": debug_mode or self.debug_mode, } + + # Check if background_tasks is available and ALL hooks should run in background + # Note: Pre-hooks running in background may not be able to modify run_input + if self._run_hooks_in_background is True and background_tasks is not None: + # Schedule ALL pre_hooks as background tasks + # Copy args to prevent race conditions + bg_args = copy_args_for_background(all_args) + for hook in hooks: + # Filter arguments to only include those that the hook accepts + filtered_args = filter_hook_args(hook, bg_args) + + # Add to background tasks + background_tasks.add_task(hook, **filtered_args) + return + all_args.update(kwargs) for i, hook in enumerate(hooks): + # Check if this specific hook should run in background (via @hook decorator) + if should_run_hook_in_background(hook) and background_tasks is not None: + # Copy args to prevent race conditions + bg_args = copy_args_for_background(all_args) + filtered_args = filter_hook_args(hook, bg_args) + background_tasks.add_task(hook, **filtered_args) + continue + if stream_events: yield handle_event( # type: ignore run_response=run_response, @@ -4125,13 +4199,13 @@ class Agent: user_id: Optional[str] = None, debug_mode: Optional[bool] = None, stream_events: bool = False, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> AsyncIterator[RunOutputEvent]: """Execute multiple pre-hook functions in succession (async version).""" if hooks is None: return - - # Prepare all possible arguments once + # Prepare arguments for this hook all_args = { "run_input": run_input, "agent": self, @@ -4143,9 +4217,32 @@ class Agent: "user_id": user_id, "debug_mode": debug_mode or self.debug_mode, } + + # Check if background_tasks is available and ALL hooks should run in background + # Note: Pre-hooks running in background may not be able to modify run_input + if self._run_hooks_in_background is True and background_tasks is not None: + # Schedule ALL pre_hooks as background tasks + # Copy args to prevent race conditions + bg_args = copy_args_for_background(all_args) + for hook in hooks: + # Filter arguments to only include those that the hook accepts + filtered_args = filter_hook_args(hook, bg_args) + + # Add to background tasks (both sync and async hooks supported) + background_tasks.add_task(hook, **filtered_args) + return + all_args.update(kwargs) for i, hook in enumerate(hooks): + # Check if this specific hook should run in background (via @hook decorator) + if should_run_hook_in_background(hook) and background_tasks is not None: + # Copy args to prevent race conditions + bg_args = copy_args_for_background(all_args) + filtered_args = filter_hook_args(hook, bg_args) + background_tasks.add_task(hook, **filtered_args) + continue + if stream_events: yield handle_event( # type: ignore run_response=run_response, @@ -4200,13 +4297,14 @@ class Agent: user_id: Optional[str] = None, debug_mode: Optional[bool] = None, stream_events: bool = False, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> Iterator[RunOutputEvent]: """Execute multiple post-hook functions in succession.""" if hooks is None: return - # Prepare all possible arguments once + # Prepare arguments for this hook all_args = { "run_output": run_output, "agent": self, @@ -4218,9 +4316,31 @@ class Agent: "run_context": run_context, "debug_mode": debug_mode or self.debug_mode, } + + # Check if background_tasks is available and ALL hooks should run in background + if self._run_hooks_in_background is True and background_tasks is not None: + # Schedule ALL post_hooks as background tasks + # Copy args to prevent race conditions + bg_args = copy_args_for_background(all_args) + for hook in hooks: + # Filter arguments to only include those that the hook accepts + filtered_args = filter_hook_args(hook, bg_args) + + # Add to background tasks + background_tasks.add_task(hook, **filtered_args) + return + all_args.update(kwargs) for i, hook in enumerate(hooks): + # Check if this specific hook should run in background (via @hook decorator) + if should_run_hook_in_background(hook) and background_tasks is not None: + # Copy args to prevent race conditions + bg_args = copy_args_for_background(all_args) + filtered_args = filter_hook_args(hook, bg_args) + background_tasks.add_task(hook, **filtered_args) + continue + if stream_events: yield handle_event( # type: ignore run_response=run_output, @@ -4265,13 +4385,14 @@ class Agent: user_id: Optional[str] = None, debug_mode: Optional[bool] = None, stream_events: bool = False, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> AsyncIterator[RunOutputEvent]: """Execute multiple post-hook functions in succession (async version).""" if hooks is None: return - # Prepare all possible arguments once + # Prepare arguments for this hook all_args = { "run_output": run_output, "agent": self, @@ -4283,9 +4404,29 @@ class Agent: "user_id": user_id, "debug_mode": debug_mode or self.debug_mode, } + # Check if background_tasks is available and ALL hooks should run in background + if self._run_hooks_in_background is True and background_tasks is not None: + # Copy args to prevent race conditions + bg_args = copy_args_for_background(all_args) + for hook in hooks: + # Filter arguments to only include those that the hook accepts + filtered_args = filter_hook_args(hook, bg_args) + + # Add to background tasks (both sync and async hooks supported) + background_tasks.add_task(hook, **filtered_args) + return + all_args.update(kwargs) for i, hook in enumerate(hooks): + # Check if this specific hook should run in background (via @hook decorator) + if should_run_hook_in_background(hook) and background_tasks is not None: + # Copy args to prevent race conditions + bg_args = copy_args_for_background(all_args) + filtered_args = filter_hook_args(hook, bg_args) + background_tasks.add_task(hook, **filtered_args) + continue + if stream_events: yield handle_event( # type: ignore run_response=run_output, @@ -8668,6 +8809,7 @@ class Agent: # Update fields if provided if update: fields_for_new_agent.update(update) + # Create a new Agent new_agent = self.__class__(**fields_for_new_agent) log_debug(f"Created new {self.__class__.__name__}") diff --git a/libs/agno/agno/eval/accuracy.py b/libs/agno/agno/eval/accuracy.py index 29b0ad7a73..edfee36b3d 100644 --- a/libs/agno/agno/eval/accuracy.py +++ b/libs/agno/agno/eval/accuracy.py @@ -612,11 +612,14 @@ Remember: You must only compare the agent_output to the expected_output. The exp print_results: bool = True, ) -> Optional[AccuracyResult]: """Run the evaluation logic against the given answer, instead of generating an answer with the Agent""" + # Generate unique run_id for this execution (don't modify self.eval_id due to concurrency) + run_id = str(uuid4()) + set_log_level_to_debug() if self.debug_mode else set_log_level_to_info() self.result = AccuracyResult() - logger.debug(f"************ Evaluation Start: {self.eval_id} ************") + logger.debug(f"************ Evaluation Start: {run_id} ************") evaluator_agent = self.get_evaluator_agent() eval_input = self.get_eval_input() @@ -721,7 +724,7 @@ Remember: You must only compare the agent_output to the expected_output. The exp ), ) - logger.debug(f"*********** Evaluation End: {self.eval_id} ***********") + logger.debug(f"*********** Evaluation End: {run_id} ***********") return self.result async def arun_with_output( @@ -732,11 +735,14 @@ Remember: You must only compare the agent_output to the expected_output. The exp print_results: bool = True, ) -> Optional[AccuracyResult]: """Run the evaluation logic against the given answer, instead of generating an answer with the Agent""" + # Generate unique run_id for this execution (don't modify self.eval_id due to concurrency) + run_id = str(uuid4()) + set_log_level_to_debug() if self.debug_mode else set_log_level_to_info() self.result = AccuracyResult() - logger.debug(f"************ Evaluation Start: {self.eval_id} ************") + logger.debug(f"************ Evaluation Start: {run_id} ************") evaluator_agent = self.get_evaluator_agent() eval_input = self.get_eval_input() @@ -820,7 +826,7 @@ Remember: You must only compare the agent_output to the expected_output. The exp eval_input=log_eval_input, ) - logger.debug(f"*********** Evaluation End: {self.eval_id} ***********") + logger.debug(f"*********** Evaluation End: {run_id} ***********") return self.result def _get_telemetry_data(self) -> Dict[str, Any]: diff --git a/libs/agno/agno/eval/performance.py b/libs/agno/agno/eval/performance.py index baeae91ab9..39f9d6447e 100644 --- a/libs/agno/agno/eval/performance.py +++ b/libs/agno/agno/eval/performance.py @@ -498,13 +498,16 @@ class PerformanceEval: from rich.live import Live from rich.status import Status + # Generate unique run_id for this execution (don't modify self.eval_id due to concurrency) + run_id = str(uuid4()) + run_times = [] memory_usages = [] previous_snapshot = None self._set_log_level() - log_debug(f"************ Evaluation Start: {self.eval_id} ************") + log_debug(f"************ Evaluation Start: {run_id} ************") # Add a spinner while running the evaluations console = Console() @@ -615,7 +618,7 @@ class PerformanceEval: ), ) - log_debug(f"*********** Evaluation End: {self.eval_id} ***********") + log_debug(f"*********** Evaluation End: {run_id} ***********") return self.result async def arun( @@ -641,13 +644,16 @@ class PerformanceEval: from rich.live import Live from rich.status import Status + # Generate unique run_id for this execution (don't modify self.eval_id due to concurrency) + run_id = str(uuid4()) + run_times = [] memory_usages = [] previous_snapshot = None self._set_log_level() - log_debug(f"************ Evaluation Start: {self.eval_id} ************") + log_debug(f"************ Evaluation Start: {run_id} ************") # Add a spinner while running the evaluations console = Console() @@ -758,7 +764,7 @@ class PerformanceEval: ), ) - log_debug(f"*********** Evaluation End: {self.eval_id} ***********") + log_debug(f"*********** Evaluation End: {run_id} ***********") return self.result def _get_telemetry_data(self) -> Dict[str, Any]: diff --git a/libs/agno/agno/eval/reliability.py b/libs/agno/agno/eval/reliability.py index 104328504a..4390dad6f8 100644 --- a/libs/agno/agno/eval/reliability.py +++ b/libs/agno/agno/eval/reliability.py @@ -4,6 +4,7 @@ from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union from uuid import uuid4 from agno.db.base import AsyncBaseDb, BaseDb +from agno.run.team import TeamRunOutput if TYPE_CHECKING: from rich.console import Console @@ -11,7 +12,6 @@ if TYPE_CHECKING: from agno.agent import RunOutput from agno.db.schemas.evals import EvalType from agno.eval.utils import async_log_eval, log_eval_run, store_result_in_file -from agno.run.team import TeamRunOutput from agno.utils.log import logger @@ -86,6 +86,9 @@ class ReliabilityEval: from rich.live import Live from rich.status import Status + # Generate unique run_id for this execution (don't modify self.eval_id due to concurrency) + run_id = str(uuid4()) + # Add a spinner while running the evaluations console = Console() with Live(console=console, transient=True) as live_log: @@ -118,7 +121,7 @@ class ReliabilityEval: if not tool_name: continue else: - if tool_name not in self.expected_tool_calls: # type: ignore + if self.expected_tool_calls is not None and tool_name not in self.expected_tool_calls: failed_tool_calls.append(tool_call.get("function", {}).get("name")) else: passed_tool_calls.append(tool_call.get("function", {}).get("name")) @@ -183,7 +186,7 @@ class ReliabilityEval: ), ) - logger.debug(f"*********** Evaluation End: {self.eval_id} ***********") + logger.debug(f"*********** Evaluation End: {run_id} ***********") return self.result async def arun(self, *, print_results: bool = False) -> Optional[ReliabilityResult]: @@ -199,6 +202,9 @@ class ReliabilityEval: from rich.live import Live from rich.status import Status + # Generate unique run_id for this execution (don't modify self.eval_id due to concurrency) + run_id = str(uuid4()) + # Add a spinner while running the evaluations console = Console() with Live(console=console, transient=True) as live_log: @@ -223,15 +229,18 @@ class ReliabilityEval: failed_tool_calls = [] passed_tool_calls = [] - for tool_call in actual_tool_calls: # type: ignore - tool_name = tool_call.get("function", {}).get("name") - if not tool_name: - continue - else: - if tool_name not in self.expected_tool_calls: # type: ignore - failed_tool_calls.append(tool_call.get("function", {}).get("name")) + if not actual_tool_calls: + failed_tool_calls = self.expected_tool_calls or [] + else: + for tool_call in actual_tool_calls: # type: ignore + tool_name = tool_call.get("function", {}).get("name") + if not tool_name: + continue else: - passed_tool_calls.append(tool_call.get("function", {}).get("name")) + if self.expected_tool_calls is not None and tool_name not in self.expected_tool_calls: + failed_tool_calls.append(tool_call.get("function", {}).get("name")) + else: + passed_tool_calls.append(tool_call.get("function", {}).get("name")) self.result = ReliabilityResult( eval_status="PASSED" if len(failed_tool_calls) == 0 else "FAILED", @@ -244,7 +253,7 @@ class ReliabilityEval: store_result_in_file( file_path=self.file_path_to_save_results, name=self.name, - eval_id=self.eval_id, + eval_id=run_id, result=self.result, ) @@ -293,7 +302,7 @@ class ReliabilityEval: ), ) - logger.debug(f"*********** Evaluation End: {self.eval_id} ***********") + logger.debug(f"*********** Evaluation End: {run_id} ***********") return self.result def _get_telemetry_data(self) -> Dict[str, Any]: diff --git a/libs/agno/agno/hooks/__init__.py b/libs/agno/agno/hooks/__init__.py new file mode 100644 index 0000000000..f20c1ab91f --- /dev/null +++ b/libs/agno/agno/hooks/__init__.py @@ -0,0 +1,3 @@ +from agno.hooks.decorator import hook, should_run_in_background + +__all__ = ["hook", "should_run_in_background"] diff --git a/libs/agno/agno/hooks/decorator.py b/libs/agno/agno/hooks/decorator.py new file mode 100644 index 0000000000..c8c269ebfc --- /dev/null +++ b/libs/agno/agno/hooks/decorator.py @@ -0,0 +1,164 @@ +from functools import wraps +from typing import Any, Callable, TypeVar, Union, overload + +# Type variable for better type hints +F = TypeVar("F", bound=Callable[..., Any]) + +# Attribute name used to mark hooks for background execution +HOOK_RUN_IN_BACKGROUND_ATTR = "_agno_run_in_background" + + +def _is_async_function(func: Callable) -> bool: + """ + Check if a function is async, even when wrapped by decorators like @staticmethod. + Traverses the full wrapper chain to find the original function. + """ + from inspect import iscoroutinefunction, unwrap + + # First, try the standard inspect function on the wrapper + if iscoroutinefunction(func): + return True + + # Use unwrap to traverse the full __wrapped__ chain to the original function + try: + original_func = unwrap(func) + if original_func is not func and iscoroutinefunction(original_func): + return True + except ValueError: + # unwrap raises ValueError if it hits a cycle + pass + + # Check if the function has CO_COROUTINE flag in its code object + try: + if hasattr(func, "__code__") and func.__code__.co_flags & 0x80: # CO_COROUTINE flag + return True + except (AttributeError, TypeError): + pass + + return False + + +@overload +def hook() -> Callable[[F], F]: ... + + +@overload +def hook( + *, + run_in_background: bool = False, +) -> Callable[[F], F]: ... + + +@overload +def hook(func: F) -> F: ... + + +def hook(*args, **kwargs) -> Union[F, Callable[[F], F]]: + """Decorator to configure hook behavior. + + Args: + run_in_background: If True, this hook will be scheduled as a FastAPI background task + when background_tasks is available, regardless of the agent/team's + run_hooks_in_background setting. This allows per-hook control over + background execution. This is only use-able when running with AgentOS. + + Returns: + Union[F, Callable[[F], F]]: Decorated function or decorator + + Examples: + @hook + def my_hook(run_output, agent): + # This runs normally (blocking) + process_output(run_output.content) + + @hook() + def another_hook(run_output, agent): + # Same as above - runs normally + process_output(run_output.content) + + @hook(run_in_background=True) + def my_background_hook(run_output, agent): + # This will run in the background when background_tasks is available + send_notification(run_output.content) + + @hook(run_in_background=True) + async def my_async_background_hook(run_output, agent): + # Async hooks also supported + await send_async_notification(run_output.content) + + agent = Agent( + model=OpenAIChat(id="gpt-4o"), + post_hooks=[my_hook, my_background_hook], + ) + """ + # Valid kwargs for the hook decorator + VALID_KWARGS = frozenset({"run_in_background"}) + + # Validate kwargs + invalid_kwargs = set(kwargs.keys()) - VALID_KWARGS + if invalid_kwargs: + raise ValueError( + f"Invalid hook configuration arguments: {invalid_kwargs}. Valid arguments are: {sorted(VALID_KWARGS)}" + ) + + def decorator(func: F) -> F: + run_in_background = kwargs.get("run_in_background", False) + + # Preserve existing hook attributes from previously applied decorators + # Use OR logic: if any decorator sets run_in_background=True, it stays True + existing_run_in_background = should_run_in_background(func) + final_run_in_background = run_in_background or existing_run_in_background + + @wraps(func) + def sync_wrapper(*args: Any, **kwargs: Any) -> Any: + return func(*args, **kwargs) + + @wraps(func) + async def async_wrapper(*args: Any, **kwargs: Any) -> Any: + return await func(*args, **kwargs) + + # Choose appropriate wrapper based on function type + if _is_async_function(func): + wrapper = async_wrapper + else: + wrapper = sync_wrapper + + # Set the background execution attribute (combined from all decorators) + setattr(wrapper, HOOK_RUN_IN_BACKGROUND_ATTR, final_run_in_background) + + return wrapper # type: ignore + + # Handle both @hook and @hook() cases + if len(args) == 1 and callable(args[0]) and not kwargs: + return decorator(args[0]) + + return decorator + + +def should_run_in_background(hook_func: Callable) -> bool: + """ + Check if a hook function is marked to run in background. + Traverses the wrapper chain to find the attribute when multiple decorators are stacked. + + Args: + hook_func: The hook function to check + + Returns: + True if the hook is decorated with @hook(run_in_background=True) + """ + # Check the function directly first + if hasattr(hook_func, HOOK_RUN_IN_BACKGROUND_ATTR): + return getattr(hook_func, HOOK_RUN_IN_BACKGROUND_ATTR) + + # Traverse the wrapper chain to find the attribute + current = hook_func + seen: set[int] = set() + while hasattr(current, "__wrapped__"): + if id(current) in seen: + break + seen.add(id(current)) + current = current.__wrapped__ + if hasattr(current, HOOK_RUN_IN_BACKGROUND_ATTR): + return getattr(current, HOOK_RUN_IN_BACKGROUND_ATTR) + + return False diff --git a/libs/agno/agno/knowledge/chunking/semantic.py b/libs/agno/agno/knowledge/chunking/semantic.py index aa608aae4c..cc0d319c8a 100644 --- a/libs/agno/agno/knowledge/chunking/semantic.py +++ b/libs/agno/agno/knowledge/chunking/semantic.py @@ -55,10 +55,10 @@ class SemanticChunking(ChunkingStrategy): # Fallback to model id params["embedding_model"] = getattr(self.embedder, "id", None) or "text-embedding-3-small" - self.chunker = SemanticChunker(**params) + self.chunker = SemanticChunker(**params) # type: ignore except Exception: # As a final fallback, use the original behavior - self.chunker = SemanticChunker( + self.chunker = SemanticChunker( # type: ignore embedding_model=getattr(self.embedder, "id", None) or "text-embedding-3-small", chunk_size=self.chunk_size, threshold=self.similarity_threshold, diff --git a/libs/agno/agno/models/anthropic/claude.py b/libs/agno/agno/models/anthropic/claude.py index 5d083bb141..73983d01b1 100644 --- a/libs/agno/agno/models/anthropic/claude.py +++ b/libs/agno/agno/models/anthropic/claude.py @@ -933,8 +933,8 @@ class Claude(Model): elif isinstance(response, (ContentBlockStopEvent, ParsedBetaContentBlockStopEvent)): if response.content_block.type == "tool_use": # type: ignore tool_use = response.content_block # type: ignore - tool_name = tool_use.name - tool_input = tool_use.input + tool_name = tool_use.name # type: ignore + tool_input = tool_use.input # type: ignore function_def = {"name": tool_name} if tool_input: @@ -944,7 +944,7 @@ class Claude(Model): model_response.tool_calls = [ { - "id": tool_use.id, + "id": tool_use.id, # type: ignore "type": "function", "function": function_def, } @@ -965,7 +965,7 @@ class Claude(Model): for block in response.message.content: # type: ignore # Handle text blocks for structured output parsing if block.type == "text": - accumulated_text += block.text + accumulated_text += block.text # type: ignore # Handle citations citations = getattr(block, "citations", None) diff --git a/libs/agno/agno/os/app.py b/libs/agno/agno/os/app.py index 1378b8fbb3..9a8c48b4be 100644 --- a/libs/agno/agno/os/app.py +++ b/libs/agno/agno/os/app.py @@ -110,6 +110,7 @@ class AgentOS: on_route_conflict: Literal["preserve_agentos", "preserve_base_app", "error"] = "preserve_agentos", telemetry: bool = True, auto_provision_dbs: bool = True, + run_hooks_in_background: bool = False, ): """Initialize AgentOS. @@ -131,6 +132,7 @@ class AgentOS: base_app: Optional base FastAPI app to use for the AgentOS. All routes and middleware will be added to this app. on_route_conflict: What to do when a route conflict is detected in case a custom base_app is provided. telemetry: Whether to enable telemetry + run_hooks_in_background: If True, run agent/team pre/post hooks as FastAPI background tasks (non-blocking) """ if not agents and not workflows and not teams and not knowledge: @@ -173,6 +175,9 @@ class AgentOS: self.enable_mcp_server = enable_mcp_server self.lifespan = lifespan + # If True, run agent/team hooks as FastAPI background tasks + self.run_hooks_in_background = run_hooks_in_background + # List of all MCP tools used inside the AgentOS self.mcp_tools: List[Any] = [] self._mcp_app: Optional[Any] = None @@ -332,6 +337,9 @@ class AgentOS: # Required for the built-in routes to work agent.store_events = True + # Propagate run_hooks_in_background setting from AgentOS to agents + agent._run_hooks_in_background = self.run_hooks_in_background + def _initialize_teams(self) -> None: """Initialize and configure all teams for AgentOS usage.""" if not self.teams: @@ -353,6 +361,9 @@ class AgentOS: # Required for the built-in routes to work team.store_events = True + # Propagate run_hooks_in_background setting to team and all nested members + team.propagate_run_hooks_in_background(self.run_hooks_in_background) + def _initialize_workflows(self) -> None: """Initialize and configure all workflows for AgentOS usage.""" if not self.workflows: @@ -369,6 +380,9 @@ class AgentOS: # Required for the built-in routes to work workflow.store_events = True + # Propagate run_hooks_in_background setting to workflow and all its step agents/teams + workflow.propagate_run_hooks_in_background(self.run_hooks_in_background) + def get_app(self) -> FastAPI: if self.base_app: fastapi_app = self.base_app diff --git a/libs/agno/agno/os/router.py b/libs/agno/agno/os/router.py index 2d6bb61446..86c79e8061 100644 --- a/libs/agno/agno/os/router.py +++ b/libs/agno/agno/os/router.py @@ -4,6 +4,7 @@ from uuid import uuid4 from fastapi import ( APIRouter, + BackgroundTasks, Depends, File, Form, @@ -300,9 +301,14 @@ async def agent_response_streamer( audio: Optional[List[Audio]] = None, videos: Optional[List[Video]] = None, files: Optional[List[FileMedia]] = None, + background_tasks: Optional[BackgroundTasks] = None, **kwargs: Any, ) -> AsyncGenerator: try: + # Pass background_tasks if provided + if background_tasks is not None: + kwargs["background_tasks"] = background_tasks + run_response = agent.arun( input=message, session_id=session_id, @@ -341,6 +347,7 @@ async def agent_continue_response_streamer( updated_tools: Optional[List] = None, session_id: Optional[str] = None, user_id: Optional[str] = None, + background_tasks: Optional[BackgroundTasks] = None, ) -> AsyncGenerator: try: continue_response = agent.acontinue_run( @@ -350,6 +357,7 @@ async def agent_continue_response_streamer( user_id=user_id, stream=True, stream_events=True, + background_tasks=background_tasks, ) async for run_response_chunk in continue_response: yield format_sse_event(run_response_chunk) # type: ignore @@ -384,10 +392,15 @@ async def team_response_streamer( audio: Optional[List[Audio]] = None, videos: Optional[List[Video]] = None, files: Optional[List[FileMedia]] = None, + background_tasks: Optional[BackgroundTasks] = None, **kwargs: Any, ) -> AsyncGenerator: """Run the given team asynchronously and yield its response""" try: + # Pass background_tasks if provided + if background_tasks is not None: + kwargs["background_tasks"] = background_tasks + run_response = team.arun( input=message, session_id=session_id, @@ -494,9 +507,14 @@ async def workflow_response_streamer( input: Optional[Union[str, Dict[str, Any], List[Any], BaseModel]] = None, session_id: Optional[str] = None, user_id: Optional[str] = None, + background_tasks: Optional[BackgroundTasks] = None, **kwargs: Any, ) -> AsyncGenerator: try: + # Pass background_tasks if provided + if background_tasks is not None: + kwargs["background_tasks"] = background_tasks + run_response = workflow.arun( input=input, session_id=session_id, @@ -813,6 +831,7 @@ def get_base_router( async def create_agent_run( agent_id: str, request: Request, + background_tasks: BackgroundTasks, message: str = Form(...), stream: bool = Form(False), session_id: Optional[str] = Form(None), @@ -951,6 +970,7 @@ def get_base_router( audio=base64_audios if base64_audios else None, videos=base64_videos if base64_videos else None, files=input_files if input_files else None, + background_tasks=background_tasks, **kwargs, ), media_type="text/event-stream", @@ -968,6 +988,7 @@ def get_base_router( videos=base64_videos if base64_videos else None, files=input_files if input_files else None, stream=False, + background_tasks=background_tasks, **kwargs, ), ) @@ -1036,6 +1057,7 @@ def get_base_router( agent_id: str, run_id: str, request: Request, + background_tasks: BackgroundTasks, tools: str = Form(...), # JSON string of tools session_id: Optional[str] = Form(None), user_id: Optional[str] = Form(None), @@ -1079,6 +1101,7 @@ def get_base_router( updated_tools=updated_tools, session_id=session_id, user_id=user_id, + background_tasks=background_tasks, ), media_type="text/event-stream", ) @@ -1092,6 +1115,7 @@ def get_base_router( session_id=session_id, user_id=user_id, stream=False, + background_tasks=background_tasks, ), ) return run_response_obj.to_dict() @@ -1229,6 +1253,7 @@ def get_base_router( async def create_team_run( team_id: str, request: Request, + background_tasks: BackgroundTasks, message: str = Form(...), stream: bool = Form(True), monitor: bool = Form(True), @@ -1338,6 +1363,7 @@ def get_base_router( audio=base64_audios if base64_audios else None, videos=base64_videos if base64_videos else None, files=document_files if document_files else None, + background_tasks=background_tasks, **kwargs, ), media_type="text/event-stream", @@ -1353,6 +1379,7 @@ def get_base_router( videos=base64_videos if base64_videos else None, files=document_files if document_files else None, stream=False, + background_tasks=background_tasks, **kwargs, ) return run_response.to_dict() @@ -1683,6 +1710,7 @@ def get_base_router( async def create_workflow_run( workflow_id: str, request: Request, + background_tasks: BackgroundTasks, message: str = Form(...), stream: bool = Form(True), session_id: Optional[str] = Form(None), @@ -1734,6 +1762,7 @@ def get_base_router( input=message, session_id=session_id, user_id=user_id, + background_tasks=background_tasks, **kwargs, ), media_type="text/event-stream", @@ -1744,6 +1773,7 @@ def get_base_router( session_id=session_id, user_id=user_id, stream=False, + background_tasks=background_tasks, **kwargs, ) return run_response.to_dict() diff --git a/libs/agno/agno/team/team.py b/libs/agno/agno/team/team.py index 5d606f8ae9..8942bc1fe6 100644 --- a/libs/agno/agno/team/team.py +++ b/libs/agno/agno/team/team.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import asyncio import contextlib import json @@ -76,8 +78,8 @@ from agno.utils.agent import ( aget_session_state_util, aset_session_name_util, aupdate_session_state_util, - await_for_background_tasks, - await_for_background_tasks_stream, + await_for_open_threads, + await_for_thread_tasks_stream, collect_joint_audios, collect_joint_files, collect_joint_images, @@ -96,8 +98,8 @@ from agno.utils.agent import ( store_media_util, update_session_state_util, validate_media_object_id, - wait_for_background_tasks, - wait_for_background_tasks_stream, + wait_for_open_threads, + wait_for_thread_tasks_stream, ) from agno.utils.common import is_typed_dict, validate_typed_dict from agno.utils.events import ( @@ -122,7 +124,7 @@ from agno.utils.events import ( create_team_tool_call_started_event, handle_event, ) -from agno.utils.hooks import filter_hook_args, normalize_hooks +from agno.utils.hooks import copy_args_for_background, filter_hook_args, normalize_hooks, should_run_hook_in_background from agno.utils.knowledge import get_agentic_or_user_search_filters from agno.utils.log import ( log_debug, @@ -344,6 +346,8 @@ class Team: pre_hooks: Optional[List[Union[Callable[..., Any], BaseGuardrail]]] = None # Functions called after output is generated but before the response is returned post_hooks: Optional[List[Union[Callable[..., Any], BaseGuardrail]]] = None + # If True, run hooks as FastAPI background tasks (non-blocking). Set by AgentOS. + _run_hooks_in_background: Optional[bool] = None # --- Structured output --- # Input schema for validating input @@ -636,7 +640,7 @@ class Team: self.tool_call_limit = tool_call_limit self.tool_hooks = tool_hooks - # Initialize hooks with backward compatibility + # Initialize hooks self.pre_hooks = pre_hooks self.post_hooks = post_hooks @@ -842,6 +846,26 @@ class Team: for sub_member in member.members: member._initialize_member(sub_member, debug_mode=debug_mode) + def propagate_run_hooks_in_background(self, run_in_background: bool = True) -> None: + """ + Propagate _run_hooks_in_background setting to this team and all nested members recursively. + + This method sets _run_hooks_in_background on the team and all its members (agents and nested teams). + For nested teams, it recursively propagates the setting to their members as well. + + Args: + run_in_background: Whether hooks should run in background. Defaults to True. + """ + self._run_hooks_in_background = run_in_background + + for member in self.members: + if hasattr(member, "_run_hooks_in_background"): + member._run_hooks_in_background = run_in_background + + # If it's a nested team, recursively propagate to its members + if isinstance(member, Team): + member.propagate_run_hooks_in_background(run_in_background) + def _set_default_model(self) -> None: # Set the default model if self.model is None: @@ -1042,13 +1066,14 @@ class Team: user_id: Optional[str] = None, debug_mode: Optional[bool] = None, stream_events: bool = False, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> Iterator[TeamRunOutputEvent]: """Execute multiple pre-hook functions in succession.""" if hooks is None: return - # Prepare all possible arguments once + # Prepare arguments for hooks all_args = { "run_input": run_input, "run_context": run_context, @@ -1060,9 +1085,32 @@ class Team: "dependencies": run_context.dependencies, "debug_mode": debug_mode or self.debug_mode, } + + # Check if background_tasks is available and ALL hooks should run in background + # Note: Pre-hooks running in background may not be able to modify run_input + if self._run_hooks_in_background is True and background_tasks is not None: + # Schedule ALL pre_hooks as background tasks + # Copy args to prevent race conditions + bg_args = copy_args_for_background(all_args) + for hook in hooks: + # Filter arguments to only include those that the hook accepts + filtered_args = filter_hook_args(hook, bg_args) + + # Add to background tasks + background_tasks.add_task(hook, **filtered_args) + return + all_args.update(kwargs) for i, hook in enumerate(hooks): + # Check if this specific hook should run in background (via @hook decorator) + if should_run_hook_in_background(hook) and background_tasks is not None: + # Copy args to prevent race conditions + bg_args = copy_args_for_background(all_args) + filtered_args = filter_hook_args(hook, bg_args) + background_tasks.add_task(hook, **filtered_args) + continue + if stream_events: yield handle_event( # type: ignore run_response=run_response, @@ -1110,13 +1158,14 @@ class Team: user_id: Optional[str] = None, debug_mode: Optional[bool] = None, stream_events: bool = False, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> AsyncIterator[TeamRunOutputEvent]: """Execute multiple pre-hook functions in succession (async version).""" if hooks is None: return - # Prepare all possible arguments once + # Prepare arguments for hooks all_args = { "run_input": run_input, "run_context": run_context, @@ -1128,9 +1177,32 @@ class Team: "metadata": run_context.metadata, "debug_mode": debug_mode or self.debug_mode, } + + # Check if background_tasks is available and ALL hooks should run in background + # Note: Pre-hooks running in background may not be able to modify run_input + if self._run_hooks_in_background is True and background_tasks is not None: + # Schedule ALL pre_hooks as background tasks + # Copy args to prevent race conditions + bg_args = copy_args_for_background(all_args) + for hook in hooks: + # Filter arguments to only include those that the hook accepts + filtered_args = filter_hook_args(hook, bg_args) + + # Add to background tasks (both sync and async hooks supported) + background_tasks.add_task(hook, **filtered_args) + return + all_args.update(kwargs) for i, hook in enumerate(hooks): + # Check if this specific hook should run in background (via @hook decorator) + if should_run_hook_in_background(hook) and background_tasks is not None: + # Copy args to prevent race conditions + bg_args = copy_args_for_background(all_args) + filtered_args = filter_hook_args(hook, bg_args) + background_tasks.add_task(hook, **filtered_args) + continue + if stream_events: yield handle_event( # type: ignore run_response=run_response, @@ -1183,13 +1255,14 @@ class Team: user_id: Optional[str] = None, debug_mode: Optional[bool] = None, stream_events: bool = False, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> Iterator[TeamRunOutputEvent]: """Execute multiple post-hook functions in succession.""" if hooks is None: return - # Prepare all possible arguments once + # Prepare arguments for hooks all_args = { "run_output": run_output, "run_context": run_context, @@ -1201,9 +1274,31 @@ class Team: "metadata": run_context.metadata, "debug_mode": debug_mode or self.debug_mode, } + + # Check if background_tasks is available and ALL hooks should run in background + if self._run_hooks_in_background is True and background_tasks is not None: + # Schedule ALL post_hooks as background tasks + # Copy args to prevent race conditions + bg_args = copy_args_for_background(all_args) + for hook in hooks: + # Filter arguments to only include those that the hook accepts + filtered_args = filter_hook_args(hook, bg_args) + + # Add to background tasks + background_tasks.add_task(hook, **filtered_args) + return + all_args.update(kwargs) for i, hook in enumerate(hooks): + # Check if this specific hook should run in background (via @hook decorator) + if should_run_hook_in_background(hook) and background_tasks is not None: + # Copy args to prevent race conditions + bg_args = copy_args_for_background(all_args) + filtered_args = filter_hook_args(hook, bg_args) + background_tasks.add_task(hook, **filtered_args) + continue + if stream_events: yield handle_event( # type: ignore run_response=run_output, @@ -1246,13 +1341,14 @@ class Team: user_id: Optional[str] = None, debug_mode: Optional[bool] = None, stream_events: bool = False, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> AsyncIterator[TeamRunOutputEvent]: """Execute multiple post-hook functions in succession (async version).""" if hooks is None: return - # Prepare all possible arguments once + # Prepare arguments for hooks all_args = { "run_output": run_output, "run_context": run_context, @@ -1264,9 +1360,31 @@ class Team: "metadata": run_context.metadata, "debug_mode": debug_mode or self.debug_mode, } + + # Check if background_tasks is available and ALL hooks should run in background + if self._run_hooks_in_background is True and background_tasks is not None: + # Schedule ALL post_hooks as background tasks + # Copy args to prevent race conditions + bg_args = copy_args_for_background(all_args) + for hook in hooks: + # Filter arguments to only include those that the hook accepts + filtered_args = filter_hook_args(hook, bg_args) + + # Add to background tasks (both sync and async hooks supported) + background_tasks.add_task(hook, **filtered_args) + return + all_args.update(kwargs) for i, hook in enumerate(hooks): + # Check if this specific hook should run in background (via @hook decorator) + if should_run_hook_in_background(hook) and background_tasks is not None: + # Copy args to prevent race conditions + bg_args = copy_args_for_background(all_args) + filtered_args = filter_hook_args(hook, bg_args) + background_tasks.add_task(hook, **filtered_args) + continue + if stream_events: yield handle_event( # type: ignore run_response=run_output, @@ -1315,6 +1433,7 @@ class Team: add_session_state_to_context: Optional[bool] = None, response_format: Optional[Union[Dict, Type[BaseModel]]] = None, debug_mode: Optional[bool] = None, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> TeamRunOutput: """Run the Team and return the response. @@ -1351,6 +1470,7 @@ class Team: session=session, user_id=user_id, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) # Consume the generator without yielding @@ -1463,13 +1583,14 @@ class Team: session=session, user_id=user_id, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) deque(iterator, maxlen=0) raise_if_cancelled(run_response.run_id) # type: ignore # 11. Wait for background memory creation - wait_for_background_tasks(memory_future=memory_future) + wait_for_open_threads(memory_future=memory_future) raise_if_cancelled(run_response.run_id) # type: ignore @@ -1522,6 +1643,7 @@ class Team: stream_events: bool = False, yield_run_output: bool = False, debug_mode: Optional[bool] = None, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> Iterator[Union[TeamRunOutputEvent, RunOutputEvent, TeamRunOutput]]: """Run the Team and return the response iterator. @@ -1555,6 +1677,7 @@ class Team: user_id=user_id, debug_mode=debug_mode, stream_events=stream_events, + background_tasks=background_tasks, **kwargs, ) for pre_hook_event in pre_hook_iterator: @@ -1707,12 +1830,13 @@ class Team: user_id=user_id, debug_mode=debug_mode, stream_events=stream_events, + background_tasks=background_tasks, **kwargs, ) raise_if_cancelled(run_response.run_id) # type: ignore # 8. Wait for background memory creation - yield from wait_for_background_tasks_stream( + yield from wait_for_thread_tasks_stream( run_response=run_response, memory_future=memory_future, stream_events=stream_events, @@ -1899,6 +2023,12 @@ class Team: stacklevel=2, ) + background_tasks = kwargs.pop("background_tasks", None) + if background_tasks is not None: + from fastapi import BackgroundTasks + + background_tasks: BackgroundTasks = background_tasks # type: ignore + # Create a run_id for this specific run run_id = str(uuid4()) @@ -2056,6 +2186,7 @@ class Team: stream_events=stream_events, yield_run_output=yield_run_output, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) @@ -2071,6 +2202,7 @@ class Team: add_session_state_to_context=add_session_state, response_format=response_format, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) @@ -2128,6 +2260,7 @@ class Team: add_session_state_to_context: Optional[bool] = None, add_history_to_context: Optional[bool] = None, debug_mode: Optional[bool] = None, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> TeamRunOutput: """Run the Team and return the response. @@ -2189,6 +2322,7 @@ class Team: session=team_session, user_id=user_id, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) @@ -2304,6 +2438,7 @@ class Team: session=team_session, user_id=user_id, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ): pass @@ -2311,7 +2446,7 @@ class Team: raise_if_cancelled(run_response.run_id) # type: ignore # 13. Wait for background memory creation - await await_for_background_tasks(memory_task=memory_task) + await await_for_open_threads(memory_task=memory_task) raise_if_cancelled(run_response.run_id) # type: ignore # 14. Create session summary @@ -2372,6 +2507,7 @@ class Team: add_session_state_to_context: Optional[bool] = None, add_history_to_context: Optional[bool] = None, debug_mode: Optional[bool] = None, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> AsyncIterator[Union[TeamRunOutputEvent, RunOutputEvent, TeamRunOutput]]: """Run the Team and return the response. @@ -2430,6 +2566,7 @@ class Team: user_id=user_id, debug_mode=debug_mode, stream_events=stream_events, + background_tasks=background_tasks, **kwargs, ) async for pre_hook_event in pre_hook_iterator: @@ -2584,13 +2721,14 @@ class Team: user_id=user_id, debug_mode=debug_mode, stream_events=stream_events, + background_tasks=background_tasks, **kwargs, ): yield event raise_if_cancelled(run_response.run_id) # type: ignore # 11. Wait for background memory creation - async for event in await_for_background_tasks_stream( + async for event in await_for_thread_tasks_stream( run_response=run_response, memory_task=memory_task, stream_events=stream_events, @@ -2784,6 +2922,12 @@ class Team: stacklevel=2, ) + background_tasks = kwargs.pop("background_tasks", None) + if background_tasks is not None: + from fastapi import BackgroundTasks + + background_tasks: BackgroundTasks = background_tasks # type: ignore + # Create a run_id for this specific run run_id = str(uuid4()) @@ -2932,6 +3076,7 @@ class Team: stream_events=stream_events, yield_run_output=yield_run_output, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) return response_iterator # type: ignore @@ -2947,6 +3092,7 @@ class Team: add_session_state_to_context=add_session_state, response_format=response_format, debug_mode=debug_mode, + background_tasks=background_tasks, **kwargs, ) diff --git a/libs/agno/agno/tools/nano_banana.py b/libs/agno/agno/tools/nano_banana.py index 68e6b51235..b8d7b1be01 100644 --- a/libs/agno/agno/tools/nano_banana.py +++ b/libs/agno/agno/tools/nano_banana.py @@ -79,7 +79,7 @@ class NanoBananaTools(Toolkit): response = client.models.generate_content( model=self.model, - contents=[prompt], + contents=[prompt], # type: ignore config=cfg, ) diff --git a/libs/agno/agno/utils/agent.py b/libs/agno/agno/utils/agent.py index c59ba5f434..78d8b407cb 100644 --- a/libs/agno/agno/utils/agent.py +++ b/libs/agno/agno/utils/agent.py @@ -24,7 +24,7 @@ if TYPE_CHECKING: from agno.team.team import Team -async def await_for_background_tasks( +async def await_for_open_threads( memory_task: Optional[Task] = None, cultural_knowledge_task: Optional[Task] = None, ) -> None: @@ -41,7 +41,7 @@ async def await_for_background_tasks( log_warning(f"Error in cultural knowledge creation: {str(e)}") -def wait_for_background_tasks( +def wait_for_open_threads( memory_future: Optional[Future] = None, cultural_knowledge_future: Optional[Future] = None ) -> None: if memory_future is not None: @@ -58,7 +58,7 @@ def wait_for_background_tasks( log_warning(f"Error in cultural knowledge creation: {str(e)}") -async def await_for_background_tasks_stream( +async def await_for_thread_tasks_stream( run_response: Union[RunOutput, TeamRunOutput], memory_task: Optional[Task] = None, cultural_knowledge_task: Optional[Task] = None, @@ -109,7 +109,7 @@ async def await_for_background_tasks_stream( log_warning(f"Error in cultural knowledge creation: {str(e)}") -def wait_for_background_tasks_stream( +def wait_for_thread_tasks_stream( run_response: Union[TeamRunOutput, RunOutput], memory_future: Optional[Future] = None, cultural_knowledge_future: Optional[Future] = None, diff --git a/libs/agno/agno/utils/hooks.py b/libs/agno/agno/utils/hooks.py index c1eb12a03e..bff29a4a32 100644 --- a/libs/agno/agno/utils/hooks.py +++ b/libs/agno/agno/utils/hooks.py @@ -1,14 +1,69 @@ +from copy import deepcopy from typing import Any, Callable, Dict, List, Optional, Union from agno.guardrails.base import BaseGuardrail +from agno.hooks.decorator import HOOK_RUN_IN_BACKGROUND_ATTR from agno.utils.log import log_warning +# Keys that should be deep copied for background hooks to prevent race conditions +BACKGROUND_HOOK_COPY_KEYS = frozenset( + {"run_input", "run_context", "run_output", "session_state", "dependencies", "metadata"} +) + + +def copy_args_for_background(args: Dict[str, Any]) -> Dict[str, Any]: + """ + Create a copy of hook arguments for background execution. + + This deep copies run_input, run_context, run_output, session_state, dependencies, + and metadata to prevent race conditions when hooks run in the background. + + Args: + args: The original arguments dictionary + + Returns: + A new dictionary with copied values for sensitive keys + """ + copied_args = {} + for key, value in args.items(): + if key in BACKGROUND_HOOK_COPY_KEYS and value is not None: + try: + copied_args[key] = deepcopy(value) + except Exception: + # If deepcopy fails (e.g., for non-copyable objects), use the original + log_warning(f"Could not deepcopy {key} for background hook, using original reference") + copied_args[key] = value + else: + copied_args[key] = value + return copied_args + + +def should_run_hook_in_background(hook: Callable[..., Any]) -> bool: + """ + Check if a hook function should run in background. + + This checks for the _agno_run_in_background attribute set by the @hook decorator. + + Args: + hook: The hook function to check + + Returns: + True if the hook is decorated with @hook(run_in_background=True) + """ + return getattr(hook, HOOK_RUN_IN_BACKGROUND_ATTR, False) + def normalize_hooks( hooks: Optional[List[Union[Callable[..., Any], BaseGuardrail]]], async_mode: bool = False, ) -> Optional[List[Callable[..., Any]]]: - """Normalize hooks to a list format""" + """Normalize hooks to a list format + + Args: + hooks: List of hook functions or hook instances + async_mode: Whether to use async versions of methods + """ + result_hooks: List[Callable[..., Any]] = [] if hooks is not None: diff --git a/libs/agno/agno/vectordb/qdrant/qdrant.py b/libs/agno/agno/vectordb/qdrant/qdrant.py index a8d60f1c9b..669cc02d66 100644 --- a/libs/agno/agno/vectordb/qdrant/qdrant.py +++ b/libs/agno/agno/vectordb/qdrant/qdrant.py @@ -545,13 +545,13 @@ class Qdrant(VectorDb): log_warning("Filters Expressions are not supported in Qdrant. No filters will be applied.") filters = None - filters = self._format_filters(filters or {}) # type: ignore + formatted_filters = self._format_filters(filters or {}) # type: ignore if self.search_type == SearchType.vector: - results = self._run_vector_search_sync(query, limit, filters) # type: ignore + results = self._run_vector_search_sync(query, limit, formatted_filters=formatted_filters) # type: ignore elif self.search_type == SearchType.keyword: - results = self._run_keyword_search_sync(query, limit, filters) # type: ignore + results = self._run_keyword_search_sync(query, limit, formatted_filters=formatted_filters) # type: ignore elif self.search_type == SearchType.hybrid: - results = self._run_hybrid_search_sync(query, limit, filters) # type: ignore + results = self._run_hybrid_search_sync(query, limit, formatted_filters=formatted_filters) # type: ignore else: raise ValueError(f"Unsupported search type: {self.search_type}") @@ -564,13 +564,13 @@ class Qdrant(VectorDb): log_warning("Filters Expressions are not supported in Qdrant. No filters will be applied.") filters = None - filters = self._format_filters(filters or {}) # type: ignore + formatted_filters = self._format_filters(filters or {}) # type: ignore if self.search_type == SearchType.vector: - results = await self._run_vector_search_async(query, limit, filters) # type: ignore + results = await self._run_vector_search_async(query, limit, formatted_filters=formatted_filters) # type: ignore elif self.search_type == SearchType.keyword: - results = await self._run_keyword_search_async(query, limit, filters) # type: ignore + results = await self._run_keyword_search_async(query, limit, formatted_filters=formatted_filters) # type: ignore elif self.search_type == SearchType.hybrid: - results = await self._run_hybrid_search_async(query, limit, filters) # type: ignore + results = await self._run_hybrid_search_async(query, limit, formatted_filters=formatted_filters) # type: ignore else: raise ValueError(f"Unsupported search type: {self.search_type}") @@ -580,7 +580,7 @@ class Qdrant(VectorDb): self, query: str, limit: int, - filters: Optional[Union[Dict[str, Any], List[FilterExpr]]], + formatted_filters: Optional[models.Filter], ) -> List[models.ScoredPoint]: dense_embedding = self.embedder.get_embedding(query) sparse_embedding = next(iter(self.sparse_encoder.embed([query]))).as_object() @@ -598,7 +598,7 @@ class Qdrant(VectorDb): with_vectors=True, with_payload=True, limit=limit, - query_filter=filters, + query_filter=formatted_filters, ) return call.points @@ -606,7 +606,7 @@ class Qdrant(VectorDb): self, query: str, limit: int, - filters: Optional[Union[Dict[str, Any], List[FilterExpr]]], + formatted_filters: Optional[models.Filter], ) -> List[models.ScoredPoint]: dense_embedding = self.embedder.get_embedding(query) @@ -618,7 +618,7 @@ class Qdrant(VectorDb): with_vectors=True, with_payload=True, limit=limit, - query_filter=filters, + query_filter=formatted_filters, using=self.dense_vector_name, ) else: @@ -629,7 +629,7 @@ class Qdrant(VectorDb): with_vectors=True, with_payload=True, limit=limit, - query_filter=filters, + query_filter=formatted_filters, ) return call.points @@ -637,7 +637,7 @@ class Qdrant(VectorDb): self, query: str, limit: int, - filters: Optional[Union[Dict[str, Any], List[FilterExpr]]], + formatted_filters: Optional[models.Filter], ) -> List[models.ScoredPoint]: sparse_embedding = next(iter(self.sparse_encoder.embed([query]))).as_object() call = self.client.query_points( @@ -647,7 +647,7 @@ class Qdrant(VectorDb): with_payload=True, limit=limit, using=self.sparse_vector_name, - query_filter=filters, + query_filter=formatted_filters, ) return call.points @@ -655,7 +655,7 @@ class Qdrant(VectorDb): self, query: str, limit: int, - filters: Optional[Dict[str, Any]], + formatted_filters: Optional[models.Filter], ) -> List[models.ScoredPoint]: dense_embedding = self.embedder.get_embedding(query) @@ -667,7 +667,7 @@ class Qdrant(VectorDb): with_vectors=True, with_payload=True, limit=limit, - query_filter=filters, + query_filter=formatted_filters, using=self.dense_vector_name, ) else: @@ -678,7 +678,7 @@ class Qdrant(VectorDb): with_vectors=True, with_payload=True, limit=limit, - query_filter=filters, + query_filter=formatted_filters, ) return call.points @@ -686,7 +686,7 @@ class Qdrant(VectorDb): self, query: str, limit: int, - filters: Optional[Dict[str, Any]], + formatted_filters: Optional[models.Filter], ) -> List[models.ScoredPoint]: sparse_embedding = next(iter(self.sparse_encoder.embed([query]))).as_object() call = await self.async_client.query_points( @@ -696,7 +696,7 @@ class Qdrant(VectorDb): with_payload=True, limit=limit, using=self.sparse_vector_name, - query_filter=filters, + query_filter=formatted_filters, ) return call.points @@ -704,7 +704,7 @@ class Qdrant(VectorDb): self, query: str, limit: int, - filters: Optional[Union[Dict[str, Any], List[FilterExpr]]], + formatted_filters: Optional[models.Filter], ) -> List[models.ScoredPoint]: dense_embedding = self.embedder.get_embedding(query) sparse_embedding = next(iter(self.sparse_encoder.embed([query]))).as_object() @@ -722,7 +722,7 @@ class Qdrant(VectorDb): with_vectors=True, with_payload=True, limit=limit, - query_filter=filters, + query_filter=formatted_filters, ) return call.points diff --git a/libs/agno/agno/workflow/condition.py b/libs/agno/agno/workflow/condition.py index b1f6934653..b3eadc5c8f 100644 --- a/libs/agno/agno/workflow/condition.py +++ b/libs/agno/agno/workflow/condition.py @@ -182,6 +182,7 @@ class Condition: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> StepOutput: """Execute the condition and its steps with sequential chaining if condition is true""" log_debug(f"Condition Start: {self.name}", center=True, symbol="-") @@ -226,6 +227,7 @@ class Condition: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ) # Handle both single StepOutput and List[StepOutput] (from Loop/Condition/Router steps) @@ -298,6 +300,7 @@ class Condition: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> Iterator[Union[WorkflowRunOutputEvent, StepOutput]]: """Execute the condition with streaming support - mirrors Loop logic""" log_debug(f"Condition Start: {self.name}", center=True, symbol="-") @@ -387,6 +390,7 @@ class Condition: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ): if isinstance(event, StepOutput): step_outputs_for_step.append(event) @@ -471,6 +475,7 @@ class Condition: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> StepOutput: """Async execute the condition and its steps with sequential chaining""" log_debug(f"Condition Start: {self.name}", center=True, symbol="-") @@ -516,6 +521,7 @@ class Condition: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ) # Handle both single StepOutput and List[StepOutput] @@ -586,6 +592,7 @@ class Condition: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> AsyncIterator[Union[WorkflowRunOutputEvent, TeamRunOutputEvent, RunOutputEvent, StepOutput]]: """Async execute the condition with streaming support - mirrors Loop logic""" log_debug(f"Condition Start: {self.name}", center=True, symbol="-") @@ -677,6 +684,7 @@ class Condition: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ): if isinstance(event, StepOutput): step_outputs_for_step.append(event) diff --git a/libs/agno/agno/workflow/loop.py b/libs/agno/agno/workflow/loop.py index b916924c9f..6b3e58203b 100644 --- a/libs/agno/agno/workflow/loop.py +++ b/libs/agno/agno/workflow/loop.py @@ -139,6 +139,7 @@ class Loop: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> StepOutput: """Execute loop steps with iteration control - mirrors workflow execution logic""" # Use workflow logger for loop orchestration @@ -168,6 +169,7 @@ class Loop: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ) # Handle both single StepOutput and List[StepOutput] (from Loop/Condition steps) @@ -242,6 +244,7 @@ class Loop: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> Iterator[Union[WorkflowRunOutputEvent, StepOutput]]: """Execute loop steps with streaming support - mirrors workflow execution logic""" log_debug(f"Loop Start: {self.name}", center=True, symbol="=") @@ -328,6 +331,7 @@ class Loop: add_workflow_history_to_steps=add_workflow_history_to_steps, workflow_session=workflow_session, num_history_runs=num_history_runs, + background_tasks=background_tasks, ): if isinstance(event, StepOutput): step_outputs_for_iteration.append(event) @@ -445,6 +449,7 @@ class Loop: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> StepOutput: """Execute loop steps asynchronously with iteration control - mirrors workflow execution logic""" # Use workflow logger for async loop orchestration @@ -476,6 +481,7 @@ class Loop: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ) # Handle both single StepOutput and List[StepOutput] (from Loop/Condition steps) @@ -553,6 +559,7 @@ class Loop: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> AsyncIterator[Union[WorkflowRunOutputEvent, TeamRunOutputEvent, RunOutputEvent, StepOutput]]: """Execute loop steps with async streaming support - mirrors workflow execution logic""" log_debug(f"Loop Start: {self.name}", center=True, symbol="=") @@ -639,6 +646,7 @@ class Loop: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ): if isinstance(event, StepOutput): step_outputs_for_iteration.append(event) diff --git a/libs/agno/agno/workflow/parallel.py b/libs/agno/agno/workflow/parallel.py index bc995a9674..5f5f22daeb 100644 --- a/libs/agno/agno/workflow/parallel.py +++ b/libs/agno/agno/workflow/parallel.py @@ -207,6 +207,7 @@ class Parallel: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> StepOutput: """Execute all steps in parallel and return aggregated result""" # Use workflow logger for parallel orchestration @@ -244,6 +245,7 @@ class Parallel: num_history_runs=num_history_runs, run_context=run_context, session_state=step_session_state, + background_tasks=background_tasks, ) # type: ignore[union-attr] return idx, step_result, step_session_state except Exception as exc: @@ -336,6 +338,7 @@ class Parallel: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> Iterator[Union[WorkflowRunOutputEvent, StepOutput]]: """Execute all steps in parallel with streaming support""" log_debug(f"Parallel Start: {self.name} ({len(self.steps)} steps)", center=True, symbol="=") @@ -418,6 +421,7 @@ class Parallel: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ): # Put event immediately in queue event_queue.put(("event", idx, event)) @@ -533,6 +537,7 @@ class Parallel: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> StepOutput: """Execute all steps in parallel using asyncio and return aggregated result""" # Use workflow logger for async parallel orchestration @@ -569,6 +574,7 @@ class Parallel: add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, session_state=step_session_state, + background_tasks=background_tasks, ) # type: ignore[union-attr] return idx, inner_step_result, step_session_state except Exception as exc: @@ -662,6 +668,7 @@ class Parallel: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> AsyncIterator[Union[WorkflowRunOutputEvent, TeamRunOutputEvent, RunOutputEvent, StepOutput]]: """Execute all steps in parallel with async streaming support""" log_debug(f"Parallel Start: {self.name} ({len(self.steps)} steps)", center=True, symbol="=") @@ -745,6 +752,7 @@ class Parallel: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ): # type: ignore[union-attr] # Yield events immediately to the queue await event_queue.put(("event", idx, event)) diff --git a/libs/agno/agno/workflow/router.py b/libs/agno/agno/workflow/router.py index 23471d925f..31954547df 100644 --- a/libs/agno/agno/workflow/router.py +++ b/libs/agno/agno/workflow/router.py @@ -179,6 +179,7 @@ class Router: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> StepOutput: """Execute the router and its selected steps with sequential chaining""" log_debug(f"Router Start: {self.name}", center=True, symbol="-") @@ -220,6 +221,7 @@ class Router: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ) # Handle both single StepOutput and List[StepOutput] @@ -285,6 +287,7 @@ class Router: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> Iterator[Union[WorkflowRunOutputEvent, StepOutput]]: """Execute the router with streaming support""" log_debug(f"Router Start: {self.name}", center=True, symbol="-") @@ -364,6 +367,7 @@ class Router: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ): if isinstance(event, StepOutput): step_outputs_for_step.append(event) @@ -449,6 +453,7 @@ class Router: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> StepOutput: """Async execute the router and its selected steps with sequential chaining""" log_debug(f"Router Start: {self.name}", center=True, symbol="-") @@ -491,6 +496,7 @@ class Router: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ) # Handle both single StepOutput and List[StepOutput] if isinstance(step_output, list): @@ -558,6 +564,7 @@ class Router: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> AsyncIterator[Union[WorkflowRunOutputEvent, TeamRunOutputEvent, RunOutputEvent, StepOutput]]: """Async execute the router with streaming support""" log_debug(f"Router Start: {self.name}", center=True, symbol="-") @@ -639,6 +646,7 @@ class Router: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ): if isinstance(event, StepOutput): step_outputs_for_step.append(event) diff --git a/libs/agno/agno/workflow/step.py b/libs/agno/agno/workflow/step.py index d6968f5aec..b7136f7113 100644 --- a/libs/agno/agno/workflow/step.py +++ b/libs/agno/agno/workflow/step.py @@ -229,6 +229,7 @@ class Step: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> StepOutput: """Execute the step with StepInput, returning final StepOutput (non-streaming)""" log_debug(f"Executing step: {self.name}") @@ -348,6 +349,10 @@ class Step: if isinstance(self.active_executor, Team): kwargs["store_member_responses"] = True + # Forward background_tasks if provided + if background_tasks is not None: + kwargs["background_tasks"] = background_tasks + num_history_runs = self.num_history_runs if self.num_history_runs else num_history_runs use_history = ( @@ -470,6 +475,7 @@ class Step: workflow_session: Optional["WorkflowSession"] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> Iterator[Union[WorkflowRunOutputEvent, StepOutput]]: """Execute the step with event-driven streaming support""" @@ -609,6 +615,10 @@ class Step: if isinstance(self.active_executor, Team): kwargs["store_member_responses"] = True + # Forward background_tasks if provided + if background_tasks is not None: + kwargs["background_tasks"] = background_tasks + num_history_runs = self.num_history_runs if self.num_history_runs else num_history_runs use_history = ( @@ -720,6 +730,7 @@ class Step: workflow_session: Optional["WorkflowSession"] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> StepOutput: """Execute the step with StepInput, returning final StepOutput (non-streaming)""" logger.info(f"Executing async step (non-streaming): {self.name}") @@ -870,6 +881,10 @@ class Step: if isinstance(self.active_executor, Team): kwargs["store_member_responses"] = True + # Forward background_tasks if provided + if background_tasks is not None: + kwargs["background_tasks"] = background_tasks + num_history_runs = self.num_history_runs if self.num_history_runs else num_history_runs use_history = ( @@ -945,6 +960,7 @@ class Step: workflow_session: Optional["WorkflowSession"] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> AsyncIterator[Union[WorkflowRunOutputEvent, StepOutput]]: """Execute the step with event-driven streaming support""" @@ -1128,6 +1144,10 @@ class Step: if isinstance(self.active_executor, Team): kwargs["store_member_responses"] = True + # Forward background_tasks if provided + if background_tasks is not None: + kwargs["background_tasks"] = background_tasks + num_history_runs = self.num_history_runs if self.num_history_runs else num_history_runs use_history = ( diff --git a/libs/agno/agno/workflow/steps.py b/libs/agno/agno/workflow/steps.py index 6413baaa7e..198c2f416c 100644 --- a/libs/agno/agno/workflow/steps.py +++ b/libs/agno/agno/workflow/steps.py @@ -126,6 +126,7 @@ class Steps: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> StepOutput: """Execute all steps in sequence and return the final result""" log_debug(f"Steps Start: {self.name} ({len(self.steps)} steps)", center=True, symbol="-") @@ -159,6 +160,7 @@ class Steps: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ) # Handle both single StepOutput and List[StepOutput] (from Loop/Condition/Router steps) @@ -221,6 +223,7 @@ class Steps: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> Iterator[Union[WorkflowRunOutputEvent, TeamRunOutputEvent, RunOutputEvent, StepOutput]]: """Execute all steps in sequence with streaming support""" log_debug(f"Steps Start: {self.name} ({len(self.steps)} steps)", center=True, symbol="-") @@ -291,6 +294,7 @@ class Steps: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ): if isinstance(event, StepOutput): step_outputs_for_step.append(event) @@ -372,6 +376,7 @@ class Steps: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> StepOutput: """Execute all steps in sequence asynchronously and return the final result""" log_debug(f"Steps Start: {self.name} ({len(self.steps)} steps)", center=True, symbol="-") @@ -405,6 +410,7 @@ class Steps: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ) # Handle both single StepOutput and List[StepOutput] (from Loop/Condition/Router steps) @@ -466,6 +472,7 @@ class Steps: workflow_session: Optional[WorkflowSession] = None, add_workflow_history_to_steps: Optional[bool] = False, num_history_runs: int = 3, + background_tasks: Optional[Any] = None, ) -> AsyncIterator[Union[WorkflowRunOutputEvent, TeamRunOutputEvent, RunOutputEvent, StepOutput]]: """Execute all steps in sequence with async streaming support""" log_debug(f"Steps Start: {self.name} ({len(self.steps)} steps)", center=True, symbol="-") @@ -536,6 +543,7 @@ class Steps: workflow_session=workflow_session, add_workflow_history_to_steps=add_workflow_history_to_steps, num_history_runs=num_history_runs, + background_tasks=background_tasks, ): if isinstance(event, StepOutput): step_outputs_for_step.append(event) diff --git a/libs/agno/agno/workflow/workflow.py b/libs/agno/agno/workflow/workflow.py index ec771c0aee..103c3ddfd6 100644 --- a/libs/agno/agno/workflow/workflow.py +++ b/libs/agno/agno/workflow/workflow.py @@ -186,6 +186,9 @@ class Workflow: # Deprecated. Use stream_events instead. stream_intermediate_steps: bool = False + # If True, run hooks as FastAPI background tasks (non-blocking). Set by AgentOS. + _run_hooks_in_background: bool = False + def __init__( self, id: Optional[str] = None, @@ -1287,6 +1290,7 @@ class Workflow: execution_input: WorkflowExecutionInput, workflow_run_response: WorkflowRunOutput, run_context: RunContext, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> WorkflowRunOutput: """Execute a specific pipeline by name synchronously""" @@ -1360,6 +1364,7 @@ class Workflow: if self.add_workflow_history_to_steps else None, num_history_runs=self.num_history_runs, + background_tasks=background_tasks, ) # Check for cancellation after step execution @@ -1460,6 +1465,7 @@ class Workflow: workflow_run_response: WorkflowRunOutput, run_context: RunContext, stream_events: bool = False, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> Iterator[WorkflowRunOutputEvent]: """Execute a specific pipeline by name with event streaming""" @@ -1557,6 +1563,7 @@ class Workflow: if self.add_workflow_history_to_steps else None, num_history_runs=self.num_history_runs, + background_tasks=background_tasks, ): raise_if_cancelled(workflow_run_response.run_id) # type: ignore @@ -1848,6 +1855,7 @@ class Workflow: execution_input: WorkflowExecutionInput, workflow_run_response: WorkflowRunOutput, run_context: RunContext, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> WorkflowRunOutput: """Execute a specific pipeline by name asynchronously""" @@ -1936,6 +1944,7 @@ class Workflow: if self.add_workflow_history_to_steps else None, num_history_runs=self.num_history_runs, + background_tasks=background_tasks, ) # Check for cancellation after step execution @@ -2036,6 +2045,7 @@ class Workflow: run_context: RunContext, stream_events: bool = False, websocket_handler: Optional[WebSocketHandler] = None, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> AsyncIterator[WorkflowRunOutputEvent]: """Execute a specific pipeline by name with event streaming""" @@ -2146,6 +2156,7 @@ class Workflow: if self.add_workflow_history_to_steps else None, num_history_runs=self.num_history_runs, + background_tasks=background_tasks, ): if workflow_run_response.run_id: raise_if_cancelled(workflow_run_response.run_id) @@ -3462,6 +3473,7 @@ class Workflow: stream_events: Optional[bool] = None, stream_intermediate_steps: Optional[bool] = None, background: Optional[bool] = False, + background_tasks: Optional[Any] = None, ) -> WorkflowRunOutput: ... @overload @@ -3480,6 +3492,7 @@ class Workflow: stream_events: Optional[bool] = None, stream_intermediate_steps: Optional[bool] = None, background: Optional[bool] = False, + background_tasks: Optional[Any] = None, ) -> Iterator[WorkflowRunOutputEvent]: ... def run( @@ -3497,6 +3510,7 @@ class Workflow: stream_events: Optional[bool] = None, stream_intermediate_steps: Optional[bool] = None, background: Optional[bool] = False, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> Union[WorkflowRunOutput, Iterator[WorkflowRunOutputEvent]]: """Execute the workflow synchronously with optional streaming""" @@ -3597,6 +3611,7 @@ class Workflow: workflow_run_response=workflow_run_response, stream_events=stream_events, run_context=run_context, + background_tasks=background_tasks, **kwargs, ) else: @@ -3605,6 +3620,7 @@ class Workflow: execution_input=inputs, # type: ignore[arg-type] workflow_run_response=workflow_run_response, run_context=run_context, + background_tasks=background_tasks, **kwargs, ) @@ -3625,6 +3641,7 @@ class Workflow: stream_intermediate_steps: Optional[bool] = None, background: Optional[bool] = False, websocket: Optional[WebSocket] = None, + background_tasks: Optional[Any] = None, ) -> WorkflowRunOutput: ... @overload @@ -3644,6 +3661,7 @@ class Workflow: stream_intermediate_steps: Optional[bool] = None, background: Optional[bool] = False, websocket: Optional[WebSocket] = None, + background_tasks: Optional[Any] = None, ) -> AsyncIterator[WorkflowRunOutputEvent]: ... def arun( # type: ignore @@ -3662,6 +3680,7 @@ class Workflow: stream_intermediate_steps: Optional[bool] = False, background: Optional[bool] = False, websocket: Optional[WebSocket] = None, + background_tasks: Optional[Any] = None, **kwargs: Any, ) -> Union[WorkflowRunOutput, AsyncIterator[WorkflowRunOutputEvent]]: """Execute the workflow synchronously with optional streaming""" @@ -3798,6 +3817,7 @@ class Workflow: files=files, session_state=session_state, run_context=run_context, + background_tasks=background_tasks, **kwargs, ) else: @@ -3810,6 +3830,7 @@ class Workflow: files=files, session_state=session_state, run_context=run_context, + background_tasks=background_tasks, **kwargs, ) @@ -4158,6 +4179,56 @@ class Workflow: if hasattr(member, "workflow_id"): member.workflow_id = self.id + def propagate_run_hooks_in_background(self, run_in_background: bool = True) -> None: + """ + Propagate _run_hooks_in_background setting to this workflow and all agents/teams in steps. + + This method sets _run_hooks_in_background on the workflow and all agents/teams + within its steps, including nested teams and their members. + + Args: + run_in_background: Whether hooks should run in background. Defaults to True. + """ + self._run_hooks_in_background = run_in_background + + if not self.steps or callable(self.steps): + return + + steps_list = self.steps.steps if isinstance(self.steps, Steps) else self.steps + + for step in steps_list: + self._propagate_hooks_to_step(step, run_in_background) + + def _propagate_hooks_to_step(self, step: Any, run_in_background: bool) -> None: + """Recursively propagate _run_hooks_in_background to a step and its nested content.""" + # Handle Step objects with active executor + if hasattr(step, "active_executor") and step.active_executor: + executor = step.active_executor + # If it's a team, use its propagation method + if hasattr(executor, "propagate_run_hooks_in_background"): + executor.propagate_run_hooks_in_background(run_in_background) + elif hasattr(executor, "_run_hooks_in_background"): + executor._run_hooks_in_background = run_in_background + + # Handle agent/team directly on step + if hasattr(step, "agent") and step.agent: + if hasattr(step.agent, "_run_hooks_in_background"): + step.agent._run_hooks_in_background = run_in_background + if hasattr(step, "team") and step.team: + # Use team's method to propagate to all nested members + if hasattr(step.team, "propagate_run_hooks_in_background"): + step.team.propagate_run_hooks_in_background(run_in_background) + elif hasattr(step.team, "_run_hooks_in_background"): + step.team._run_hooks_in_background = run_in_background + + # Handle nested primitives - check 'steps' and 'choices' attributes + for attr_name in ["steps", "choices"]: + if hasattr(step, attr_name): + attr_value = getattr(step, attr_name) + if attr_value and isinstance(attr_value, list): + for nested_step in attr_value: + self._propagate_hooks_to_step(nested_step, run_in_background) + ########################################################################### # Telemetry functions ########################################################################### diff --git a/libs/agno/tests/integration/os/test_backround_tasks.py b/libs/agno/tests/integration/os/test_backround_tasks.py new file mode 100644 index 0000000000..fcc36d3f7f --- /dev/null +++ b/libs/agno/tests/integration/os/test_backround_tasks.py @@ -0,0 +1,376 @@ +"""Integration tests for background tasks in AgentOS. + +Note on Testing Background Tasks: +When using httpx.AsyncClient with ASGITransport for testing, the ASGI transport +waits for all background tasks to complete before returning the response. This is +by design in the ASGI specification to ensure background tasks finish before app +shutdown. Therefore, we cannot test timing/non-blocking behavior directly. + +Instead, we use mocking to verify that hooks are properly added to FastAPI's +BackgroundTasks when run_hooks_in_background=True, which proves they will run +in the background in production environments. + +Note: run_hooks_in_background is configured at the AgentOS level (default=False) +and propagated to agents/teams, rather than being set on individual agents. +""" + +import asyncio +import json +import time +from typing import Dict +from unittest.mock import patch + +import httpx +import pytest +from httpx import ASGITransport + +from agno.agent.agent import Agent +from agno.models.openai import OpenAIChat +from agno.os import AgentOS +from agno.run.agent import RunOutput + + +@pytest.fixture +def execution_tracker() -> Dict[str, bool]: + """Shared state to track hook execution.""" + return { + "pre_hook_executed": False, + "post_hook_executed": False, + "async_post_hook_executed": False, + "response_returned": False, + } + + +@pytest.fixture +def agent_with_hooks(shared_db, execution_tracker): + """Create an agent with hooks (background mode is set by AgentOS).""" + + async def pre_hook_log(run_input, agent): + """Pre-hook that logs request.""" + execution_tracker["pre_hook_executed"] = True + + async def post_hook_log(run_output: RunOutput, agent: Agent): + """Post-hook that runs in background.""" + await asyncio.sleep(0.5) # Simulate some work + execution_tracker["post_hook_executed"] = True + + async def async_post_hook_log(run_output: RunOutput, agent: Agent): + """Async post-hook that runs in background.""" + await asyncio.sleep(0.5) # Simulate async work + execution_tracker["async_post_hook_executed"] = True + + return Agent( + name="background-task-agent", + id="background-task-agent-id", + model=OpenAIChat(id="gpt-4o"), + db=shared_db, + pre_hooks=[pre_hook_log], + post_hooks=[post_hook_log, async_post_hook_log], + ) + + +@pytest.fixture +def test_app_with_background(agent_with_hooks): + """Create a FastAPI app with background hooks enabled.""" + agent_os = AgentOS(agents=[agent_with_hooks], run_hooks_in_background=True) + return agent_os.get_app() + + +@pytest.mark.asyncio +async def test_background_hooks_non_streaming(test_app_with_background, agent_with_hooks, execution_tracker): + """Test that post-hooks run in background for non-streaming responses.""" + + async with httpx.AsyncClient( + transport=ASGITransport(app=test_app_with_background), base_url="http://test" + ) as client: + response = await client.post( + f"/agents/{agent_with_hooks.id}/runs", + data={"message": "Hello, world!", "stream": "false"}, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + + # Response should be returned immediately + assert response.status_code == 200 + response_json = response.json() + assert response_json["run_id"] is not None + assert response_json["agent_id"] == agent_with_hooks.id + + # Mark that response was returned + execution_tracker["response_returned"] = True + + # Pre-hooks should have executed (they always block) + assert execution_tracker["pre_hook_executed"] is True + + # Background tasks should have been scheduled but may not be complete yet + # Wait a bit for background tasks to complete + await asyncio.sleep(1.5) + + # Now verify background hooks executed + assert execution_tracker["post_hook_executed"] is True + assert execution_tracker["async_post_hook_executed"] is True + + +@pytest.mark.asyncio +async def test_background_hooks_streaming(test_app_with_background, agent_with_hooks, execution_tracker): + """Test that post-hooks run in background for streaming responses.""" + + async with httpx.AsyncClient( + transport=ASGITransport(app=test_app_with_background), base_url="http://test" + ) as client: + async with client.stream( + "POST", + f"/agents/{agent_with_hooks.id}/runs", + data={"message": "Hello, world!", "stream": "true"}, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) as response: + assert response.status_code == 200 + assert "text/event-stream" in response.headers.get("content-type", "") + + # Collect streaming chunks + chunks = [] + async for line in response.aiter_lines(): + if line.startswith("data: "): + data = line[6:] # Remove 'data: ' prefix + if data != "[DONE]": + chunks.append(json.loads(data)) + + # Verify we received data + assert len(chunks) > 0 + + # Mark that response was returned + execution_tracker["response_returned"] = True + + # Pre-hooks should have executed + assert execution_tracker["pre_hook_executed"] is True + + # Wait for background tasks to complete + await asyncio.sleep(1.5) + + # Verify background hooks executed + assert execution_tracker["post_hook_executed"] is True + assert execution_tracker["async_post_hook_executed"] is True + + +@pytest.mark.asyncio +async def test_background_hooks_are_added_as_background_tasks(agent_with_hooks): + """Test that hooks are added to FastAPI background tasks when run_hooks_in_background=True.""" + + tasks_added = [] + + async def tracked_post_hook(run_output: RunOutput): + """A post-hook that we can track.""" + tasks_added.append("tracked_post_hook") + + # Replace post_hooks with our tracked hook + agent_with_hooks.post_hooks = [tracked_post_hook] + + # Mock BackgroundTasks at the FastAPI level + original_add_task = None + + def mock_add_task(self, func, *args, **kwargs): + """Mock add_task to track what's being added.""" + tasks_added.append(func.__name__) + # Call the original to maintain functionality + if original_add_task: + return original_add_task(self, func, *args, **kwargs) + + # Patch BackgroundTasks.add_task method + from fastapi import BackgroundTasks + + original_add_task = BackgroundTasks.add_task + + with patch.object(BackgroundTasks, "add_task", mock_add_task): + # Create app after patching with background hooks enabled + agent_os = AgentOS(agents=[agent_with_hooks], run_hooks_in_background=True) + app = agent_os.get_app() + + async with httpx.AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + response = await client.post( + f"/agents/{agent_with_hooks.id}/runs", + data={"message": "Hello!", "stream": "false"}, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + + # Response should succeed + assert response.status_code == 200 + + # Verify that hooks were added as background tasks + # The hook function should have been added to tasks + assert len(tasks_added) > 0, "At least one background task should be added" + assert "tracked_post_hook" in tasks_added, "Our tracked hook should be in background tasks" + + +@pytest.mark.asyncio +async def test_background_hooks_with_hook_parameters(test_app_with_background, agent_with_hooks): + """Test that background hooks receive correct parameters.""" + + received_params = {} + + async def param_checking_hook(run_output: RunOutput, agent: Agent, session, user_id, run_context): + """Hook that checks it receives expected parameters.""" + received_params["run_output"] = run_output is not None + received_params["agent"] = agent is not None + received_params["session"] = session is not None + received_params["user_id"] = user_id + received_params["run_context"] = run_context is not None + + agent_with_hooks.post_hooks = [param_checking_hook] + + async with httpx.AsyncClient( + transport=ASGITransport(app=test_app_with_background), base_url="http://test" + ) as client: + response = await client.post( + f"/agents/{agent_with_hooks.id}/runs", + data={ + "message": "Test parameters", + "user_id": "test-user-123", + "stream": "false", + }, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + + assert response.status_code == 200 + + # Wait for background task + await asyncio.sleep(0.5) + + # Verify all expected parameters were passed + assert received_params["run_output"] is True + assert received_params["agent"] is True + assert received_params["session"] is True + assert received_params["user_id"] == "test-user-123" + assert received_params["run_context"] is True + + +@pytest.mark.asyncio +async def test_agent_without_background_mode(shared_db): + """Test that hooks execute synchronously when background mode is disabled on AgentOS.""" + + execution_tracker = {"hook_executed": False} + tasks_added = [] + + async def blocking_post_hook(run_output: RunOutput): + """Post-hook that executes synchronously.""" + execution_tracker["hook_executed"] = True + + agent = Agent( + name="blocking-agent", + id="blocking-agent-id", + model=OpenAIChat(id="gpt-4o"), + db=shared_db, + post_hooks=[blocking_post_hook], + ) + + # Mock add_task to track if hooks are added as background tasks + original_add_task = None + + def mock_add_task(self, func, *args, **kwargs): + """Mock add_task to track what's being added.""" + tasks_added.append(func.__name__) + if original_add_task: + return original_add_task(self, func, *args, **kwargs) + + from fastapi import BackgroundTasks + + original_add_task = BackgroundTasks.add_task + + with patch.object(BackgroundTasks, "add_task", mock_add_task): + # Disable background hooks at the AgentOS level + agent_os = AgentOS(agents=[agent], run_hooks_in_background=False) + app = agent_os.get_app() + + async with httpx.AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + response = await client.post( + f"/agents/{agent.id}/runs", + data={"message": "Hello!", "stream": "false"}, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + + # Response should succeed + assert response.status_code == 200 + + # Hook should have executed + assert execution_tracker["hook_executed"] is True + + # Verify that our post hook was NOT added to background tasks + # When run_hooks_in_background=False on AgentOS, hooks execute synchronously + assert "blocking_post_hook" not in tasks_added, "Hook should not be added as background task when disabled" + + +@pytest.mark.asyncio +async def test_background_hooks_with_multiple_hooks(test_app_with_background, agent_with_hooks): + """Test that multiple background hooks all execute.""" + + execution_count = {"count": 0} + + def hook1(run_output: RunOutput): + time.sleep(0.3) + execution_count["count"] += 1 + + def hook2(run_output: RunOutput): + time.sleep(0.3) + execution_count["count"] += 1 + + async def hook3(run_output: RunOutput): + await asyncio.sleep(0.3) + execution_count["count"] += 1 + + agent_with_hooks.post_hooks = [hook1, hook2, hook3] + + async with httpx.AsyncClient( + transport=ASGITransport(app=test_app_with_background), base_url="http://test" + ) as client: + response = await client.post( + f"/agents/{agent_with_hooks.id}/runs", + data={"message": "Test multiple hooks", "stream": "false"}, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + + assert response.status_code == 200 + + # Wait for all background tasks + await asyncio.sleep(1.5) + + # All three hooks should have executed + assert execution_count["count"] == 3 + + +@pytest.mark.asyncio +async def test_agentos_propagates_background_setting_to_agents(shared_db): + """Test that AgentOS correctly propagates run_hooks_in_background to agents.""" + + agent = Agent( + name="test-agent", + id="test-agent-id", + model=OpenAIChat(id="gpt-4o"), + db=shared_db, + ) + + # By default, _run_hooks_in_background should be False on the agent + assert agent._run_hooks_in_background is False + + # When AgentOS is created with run_hooks_in_background=True (default) + agent_os = AgentOS(agents=[agent], run_hooks_in_background=True) + agent_os.get_app() + + # The agent's _run_hooks_in_background should now be True + assert agent._run_hooks_in_background is True + + +@pytest.mark.asyncio +async def test_agentos_propagates_background_setting_disabled(shared_db): + """Test that AgentOS correctly propagates run_hooks_in_background=False to agents.""" + + agent = Agent( + name="test-agent", + id="test-agent-id", + model=OpenAIChat(id="gpt-4o"), + db=shared_db, + ) + + # When AgentOS is created with run_hooks_in_background=False + agent_os = AgentOS(agents=[agent], run_hooks_in_background=False) + agent_os.get_app() + + # The agent's _run_hooks_in_background should remain False + assert agent._run_hooks_in_background is False