AI Functions
Strands AI Functions is a Python library for building reliable AI-powered applications through a new abstraction: functions that behave like standard Python functions, but are evaluated by reasoning AI Agents.
AI Functions extend the expressivity of standard programming by offering developers a computational model that can solve tasks not easily expressible as traditional code. They can both leverage text generation capabilities (e.g., to write summaries or retrieve information) and dynamically generate and execute code to process inputs and return native Python objects. For example, an AI Function can load a user-uploaded file in an arbitrary format and convert it to a normalized DataFrame for use in the rest of the workflow.
Direct integration of AI agents in standard workflows is often avoided due to their non-deterministic nature and lack of assurance that instructions will be followed, which can cause cascading errors throughout the workflow. AI Functions address this through extensive use of post-conditions. Unlike traditional prompt-based approaches, which try to ensure correctness by relying on prompt engineering alone, AI Functions enforce correctness through runtime post-condition checking: users can specify explicit post-conditions that the output of any given step needs to satisfy. AI Functions will automatically initiate self-correcting loops to ensure these properties are respected, avoiding cascading errors in complex workflows.
Through AI Functions, developers can construct agentic workflows and agent graphs - including asynchronous ones - by writing and composing functions. They can build shareable libraries of robust, reusable agentic flows in exactly the same way they build software libraries today, and can use standard software development practices to collaborate on refining and ensuring the safety of each component.
Getting started¶
Prerequisites¶
- Python 3.12 or higher (Python 3.14+ recommended for all features)
- Valid credentials for a supported model provider (AWS Bedrock, OpenAI, etc.)
- (Recommended) uv to run the provided examples
Installation¶
# Using pip
pip install strands-ai-functions
# Using uv
uv add strands-ai-functions
Configure model provider¶
Strands AI Functions support various model providers. Change the model option in the examples below to use a different provider, model or authentication options. For example:
from ai_functions import ai_function
from strands.models.bedrock import BedrockModel
from strands.models.openai import OpenAIModel
# Use Claude Sonnet on Amazon Bedrock (default if `model` is not specified)
model = BedrockModel(model_id="anthropic.claude-sonnet-4-20250514-v1:0")
# Or use a different provider and model
model = OpenAIModel(client_args={"api_key": "<KEY>"}, model_id="gpt-4o")
@ai_function(model=model)
def my_function() -> None:
...
Defining AI Functions¶
AI Functions behave like a standard function, but their code is written in natural language rather than Python, and are executed by an LLM rather than a CPU. Here's a complete example:
from ai_functions import ai_function
from pydantic import BaseModel
# Define the structured output type - AI Functions can return primitive types,
# Pydantic models, or even native Python objects like DataFrames
class MeetingSummary(BaseModel):
attendees: list[str]
summary: str
action_items: list[str]
# The @ai_function decorator marks this as an AI Function
# When called, it automatically creates an agent and handles execution
@ai_function
def summarize_meeting(transcripts: str) -> MeetingSummary:
"""
Write a summary of the following meeting in less than 50 words.
<transcripts>
{transcripts}
</transcripts>
"""
# The docstring serves as the instruction template
# Use {variable} syntax to reference function arguments
if __name__ == "__main__":
transcripts = "[add your meeting transcripts here]"
# Call the AI Function like any other Python function
# The library handles agent orchestration and returns the validated result
meeting_summary = summarize_meeting(transcripts)
print("=== Meeting Summary ===")
print("Attendees: " + ", ".join(meeting_summary.attendees))
print("Summary:\n" + meeting_summary.summary)
print("Action Items:")
for action_item in meeting_summary.action_items:
print(action_item)
Configure Credentials
Configure model provider credentials before running examples. You may need to change the examples to use a different model provider.
Two ways to provide instructions¶
The instructions/prompt of an AI Function can be provided in two ways. The simplest is to specify the prompt as a docstring:
from ai_functions import ai_function
@ai_function
def translate(text: str, lang: str) -> str:
"""
Translate the text below to the following language: `{lang}`.
{text}
"""
The AI Function will interpret the docstring as template and attempt to replace the values using the provided arguments. This method however has limitations in some corner cases, for example if the docstring references a non-local variable. It also makes it difficult to construct prompts whose structure depends on the inputs.
Alternatively, we can construct the prompt inside the function and return it. In addition, the body of the function can also be used to perform input validation:
from ai_functions import ai_function
@ai_function
def translate(text: str, lang: str) -> str:
assert text, "`text` cannot be empty"
assert lang, "`lang` cannot be empty"
return f"""
Translate the text below to the following language: `{lang}`.
{text}
"""
AI Functions must define clear input and output types to ensure proper validation and execution. Internally, the AI Function will always execute the function with the provided arguments. If the function returns a string, it will be used as the prompt to the agent. Otherwise, it will fall back to interpreting the docstring as a template.
When using a Python executor (with code_execution_mode="local"), all input variables to the AI function are automatically loaded into the Python environment. This means the agent can directly reference and manipulate these variables in the generated code without needing to parse them from the prompt. For example, if you pass a DataFrame as an argument, the agent can directly call methods on it like df.head() or perform operations on it.
Post-conditions¶
A core notion of AI Functions is that programmers should not "prompt-and-pray" for the result returned by the agent to be correct. Rather, they should verify that the result satisfies the conditions required by their pipeline.
To this end, AI Functions expose post-conditions as a fundamental component in defining AI Functions. Post-conditions are functions (both standard Python functions or other AI Functions) that validate the input and provide feedback to the agent. This automatically instantiate a self-correcting feedback loop ensuring the correctness of the final return value of the function.
The following example extends the meeting summary from the Quickstart guide by adding user-defined post-conditions:
from ai_functions import ai_function, PostConditionResult
from pydantic import BaseModel
class MeetingSummary(BaseModel):
attendees: list[str]
summary: str
action_items: list[str]
# Post-conditions can be standard Python functions that raise an error if validation fails
def check_length(response: MeetingSummary):
length = len(response.summary.split())
assert length <= 50, f"Summary should be less than 50 words, but is {length} words long"
# A post-condition can also be an AI Function, since AI Functions *are* just functions
@ai_function
def check_style(response: MeetingSummary) -> PostConditionResult:
"""
Check if the summary below satisfies the following criteria:
- It must use bullet points
- It must provide the reader with the necessary context
<summary>
{response.summary}
</summary>
"""
# Now we can add the functions above as post-conditions to validate the model output
@ai_function(post_conditions=[check_length, check_style], max_attempts=5)
def summarize_meeting(transcripts: str) -> MeetingSummary:
"""
Write a summary of the following meeting in less than 50 words.
<transcripts>
{transcripts}
</transcripts>
"""
All post-conditions are checked in parallel. The agent receives a message reporting all errors, and can address all of them at the same time thus cutting on the number of iterations necessary to converge to a correct output.
Post-conditions can also return a PostConditionResult object instead of raising an error:
def check_length(response: MeetingSummary) -> PostConditionResult:
length = len(response.summary.split())
if length > 50:
return PostConditionResult(
passed=False,
message=f"Summary should be less than 50 words, but is {length} words long"
)
return PostConditionResult(passed=True)
Post-conditions are not limited to checking the answer of the agent. They can more generally enforce invariants about the state of the system after the agent's execution. The example below shows how to implement a universal data loader that validates the structure and types of the resulting DataFrame:
from ai_functions import ai_function
from pandas import DataFrame, api
# Post-condition validates the structure and data types of the returned DataFrame
def check_invoice_dataframe(df: DataFrame):
"""Post-condition: validate DataFrame structure."""
assert {'product_name', 'quantity', 'price', 'purchase_date'}.issubset(df.columns)
assert api.types.is_integer_dtype(df['quantity']), "quantity must be an integer"
assert api.types.is_float_dtype(df['price']), "price must be a float"
assert api.types.is_datetime64_any_dtype(df['purchase_date']), "purchase_date must be a datetime64"
assert not df.duplicated(subset=['product_name', 'price', 'purchase_date']).any(), \
"The combination of product_name, price, and purchase_date must be unique"
@ai_function(
post_conditions=[check_invoice_dataframe],
code_execution_mode="local",
code_executor_additional_imports=["pandas", "sqlite3"],
)
def import_invoice(path: str) -> DataFrame:
"""
The file `{path}` contains purchase logs. Extract them in a DataFrame with columns:
- product_name (str)
- quantity (int)
- price (float)
- purchase_date (datetime)
"""
# The agent will dynamically inspect the file format (JSON, CSV, SQLite, etc.)
# and generate the appropriate code to load and transform it into the required format
df = import_invoice('data/invoice.json')
print("Invoice total:", df['price'].sum())
Redundancy is Intentional
Note that we are telling the agents what format to return both in the prompt and as a post-condition which may feel redundant. However, agents are generally much more effective in responding to validation messages than they are at following the prompts. Moreover, this provides a strong guarantee that if the pipeline terminates, the returned DataFrame will have the correct structure without any need for manual inspection.
AI Function configuration¶
AI Functions use Strands Agent in the backend. Any valid option of strands.Agent (such as model, tools, system_prompt) can be passed in the decorator.
from ai_functions import ai_function
from strands_tools import file_read, file_write
from typing import Literal
@ai_function(tools=[file_read, file_write])
def summarize_file(path: str, output_path: str) -> Literal["done"]:
"""
Read the file {path} and write a summary in {output_path}.
"""
summarize_file("report.md", output_path="summary.md")
To simplify maintaining and sharing configuration between different AI Functions, we can use a AIFunctionConfig object:
from ai_functions import ai_function, AIFunctionConfig
from pandas import DataFrame
class Configs:
FAST_MODEL = AIFunctionConfig(model="global.anthropic.claude-haiku-4-5-20251001-v1:0")
DATA_ANALYSIS = AIFunctionConfig(
code_executor_additional_imports=["pandas.*", "numpy.*", "plotly.*"],
code_execution_mode="local",
)
# reuse a config
@ai_function(config=Configs.DATA_ANALYSIS)
def return_of_investment(data: DataFrame) -> DataFrame:
"""
Analyze `data` and return a DataFrame with the return of investment for each year.
"""
# keyword arguments can be used to override config arguments for this specific function
@ai_function(config=Configs.FAST_MODEL, tools=[web_search])
def websearch(topic: str) -> str:
"""
Research the following topic online and return a summary of your findings:
{topic}
"""
Python integration¶
AI Agents are usually limited working with serializable input-output types (strings, JSON-objects, ...) rather than with native objects of the programming language. AI Functions, on the other hand, aim to provide a natural extension of the programming language itself enabling new kind of programming patterns and abstractions. In particular, we optionally provide agents with a Python environment allowing them to dynamically generate code to process arbitrary input data and return native Python objects.
When using a Python executor (with code_execution_mode="local"), all input variables to the AI function are automatically loaded into the Python environment. This means the agent can directly reference and manipulate these variables in the generated code without needing to parse them from the prompt.
Consider for example a webapp that allows the user to upload an invoice in an arbitrary format (pdf, csv, json). The following snippet implements a "universal data loader" that given the path to a file inspects its content and automatically decide the appropriate processing pipeline to load the file and convert it to a DataFrame in the desired format:
from ai_functions import ai_function
from pandas import DataFrame
# code execution has to be explicitly enabled since it raises security risks
@ai_function(code_execution_mode="local")
def import_invoice(path: str) -> DataFrame:
"""
The file `{path}` contains purchase logs. Extract them in a DataFrame with columns:
- product_name (str)
- quantity (int)
- price (float)
- purchase_date (datetime)
"""
@ai_function(code_execution_mode="local")
def fuzzy_merge_products(invoice: DataFrame) -> DataFrame:
"""
Find product names that denote different versions of the same product and
merge them into a single name. Return a DataFrame with the new merged names.
"""
# Load a JSON (the agent has to inspect the JSON to understand how to map it to a DataFrame)
df = import_invoice('data/invoice.json')
print("Invoice total:", df['price'].sum())
# Load a SQLite database. The agent will dynamically check the schema and generate
# the necessary queries to read it and convert it to the desired format)
df = import_invoice('data/invoice.sqlite3')
# Merge revisions of the same product
df = fuzzy_merge_products(df)
Right now Strands AI Function support only "local" execution. This will create a local Python environment (similar to a Jupyter notebook) for the agent to use. Execution in a safe remote sandboxed interpreter is a planned extension.
Security Warning
The local execution environment attempts to restrict execution to explicitly allowed libraries and methods. However, executing Python code in a non-sandboxed environment is inherently unsafe. Please make sure you understand the risk and consider running the code inside a Docker container or other sandbox.
Async invocation and parallel workflows¶
AI Functions can be defined as either sync or async. The latter is particularly useful to define parallel workflows.
In the example below, we define a workflow to write a report on the current trends for a given stock. First, we conduct several searches in parallel. Then we use the result to write a report (see examples/stock_report.py for a more complex runnable example).
from ai_functions import ai_function
from pandas import DataFrame
from datetime import timedelta
from typing import Literal
import asyncio
@ai_function(tools=[...])
async def research_news(stock: str) -> str:
"""
Research and summarize the current news regarding the following stock: {stock}
"""
@ai_function(tools=[...])
async def research_price(stock: str, past_days: int) -> DataFrame:
"""
Use the `yfinance` Python package to retrieve the historical prices of {stock} in the last 30 days.
Return a dataframe with columns [date, price (float, price at market close)]
"""
@ai_function
def write_report(stock: str, news: str, prices: DataFrame) -> str:
"""
Write and return a HTML report on the trend of the stock {stock} in the last 30 days.
Use the provided `prices` DataFrame and the following summary of recent news:
{news}
"""
async def stock_research_workflow(stock: str):
# Run the two agents in parallel
news, prices = await asyncio.gather(research_news(stock), research_price(stock))
# Use their results to write a report
write_report(stock, news, prices)
AI Functions as Strands tools¶
AI Functions can also be used as tools by other agents to build multi-agent systems with orchestration:
@ai_function(
description="Perform multiple web searches relevant to query and returns a summary of the results",
tools=[...]
)
def websearch(query: str) -> str:
"""
Perform a web search on the following topic and return a summary of your findings.
---
{query}
"""
@ai_function(tools=[websearch])
def report_writer(topic: str) -> str:
"""
Research the following topic and write a report.
---
{topic}
"""
# AI Functions can also be used as tools in regular Strands agents:
# from strands import Agent
#
# agent = Agent(
# model="global.anthropic.claude-sonnet-4-5-20250929-v1:0",
# tools=[websearch]
# )
#
# response = agent("Research quantum computing and write a report")
Next steps¶
Now that you understand the core concepts, check out the examples on GitHub for complete, runnable examples demonstrating:
- Stock report generation with async workflows and Python integration
- Multi-agent orchestration with agents as tools
- Context management for long-running tasks with automatic summarization
- ... and more!
Each example includes detailed inline comments explaining the implementation.