# marvin
Source: https://askmarvin.ai/api-reference/marvin
# `marvin`
## Submodules
* [`agents`](marvin-agents)
* [`beta`](marvin-beta)
* [`cli`](marvin-cli)
* [`database`](marvin-database)
* [`defaults`](marvin-defaults)
* [`engine`](marvin-engine)
* [`fns`](marvin-fns)
* [`handlers`](marvin-handlers)
* [`instructions`](marvin-instructions)
* [`memory`](marvin-memory)
* [`prompts`](marvin-prompts)
* [`settings`](marvin-settings)
* [`tasks`](marvin-tasks)
* [`thread`](marvin-thread)
* [`tools`](marvin-tools)
* [`utilities`](marvin-utilities)
# actor
Source: https://askmarvin.ai/api-reference/marvin-agents-actor
# `marvin.agents.actor`
## Constants
### `T`
```python theme={null}
T = TypeVar('T')
```
## Classes
### `Actor`
```python theme={null}
class Actor(name: str, instructions: str | None = None, description: str | None = None, verbose: bool = False, prompt: str | Path)
```
**Methods:**
* **`end_turn`**
```python theme={null}
def end_turn(self, thread: Thread, result: AgentRunResult)
```
Called when the actor ends its turn.
* **`friendly_name`**
```python theme={null}
def friendly_name(self, verbose: bool = True) -> str
```
* **`get_agentlet`**
```python theme={null}
def get_agentlet(self, tools: Sequence[Callable[..., Any]], end_turn_tools: Sequence[EndTurn], active_mcp_servers: list[MCPServer] | None = None) -> pydantic_ai.Agent[Any, Any]
```
* **`get_current`**
```python theme={null}
def get_current(cls) -> Actor | None
```
Get the current actor from context.
* **`get_end_turn_tools`**
```python theme={null}
def get_end_turn_tools(self) -> list[EndTurn]
```
A list of `EndTurn` tools that this actor can use to end its turn.
* **`get_memories`**
```python theme={null}
def get_memories(self) -> list[Memory]
```
A list of memories that this actor can use during its turn.
* **`get_prompt`**
```python theme={null}
def get_prompt(self) -> str
```
* **`get_tools`**
```python theme={null}
def get_tools(self) -> list[Callable[..., Any]]
```
A list of tools that this actor can use during its turn.
* **`run`**
```python theme={null}
def run(self, instructions: str | Sequence[UserContent], result_type: type[T] = str, tools: list[Callable[..., Any]] = [], thread: Thread | str | None = None, handlers: list[Handler | AsyncHandler] | None = None, raise_on_failure: bool = True, **kwargs: Any = {}) -> Any
```
* **`run_async`**
```python theme={null}
def run_async(self, instructions: str | Sequence[UserContent], result_type: type[T] = str, tools: list[Callable[..., Any]] = [], thread: Thread | str | None = None, handlers: list[Handler | AsyncHandler] | None = None, raise_on_failure: bool = True, **kwargs: Any = {}) -> Any
```
* **`say`**
```python theme={null}
def say(self, message: str, instructions: str | None = None, thread: Thread | str | None = None)
```
Responds to a user message in a conversational way.
* **`say_async`**
```python theme={null}
def say_async(self, message: str, instructions: str | None = None, thread: Thread | str | None = None)
```
Responds to a user message in a conversational way.
* **`start_turn`**
```python theme={null}
def start_turn(self, thread: Thread)
```
Called when the actor starts its turn.
## Functions
### `get_current_actor`
```python theme={null}
def get_current_actor() -> Actor | None
```
Get the currently active actor from context.
Returns:
The current Actor instance or None if no actor is active.
***
**Parent Module:** [`agents`](marvin-agents)
# agent
Source: https://askmarvin.ai/api-reference/marvin-agents-agent
Agents for Marvin.
# `marvin.agents.agent`
Agents for Marvin.
An Agent is an entity that can process tasks and maintain state across interactions.
## Constants
### `T`
```python theme={null}
T = TypeVar('T')
```
## Classes
### `Agent`
```python theme={null}
class Agent(name: str = (lambda: random.choice(AGENT_NAMES))(), instructions: str | None = None, description: str | None = None, verbose: bool = False, prompt: str | Path = Path('agent.jinja'), tools: list[Callable[..., Any]] = (lambda: [])(), memories: list[Memory] = (lambda: [])(), mcp_servers: list[Any] = (lambda: [])(), model: KnownModelName | Model | None = None, model_settings: ModelSettings = ModelSettings())
```
An agent that can process tasks and maintain state.
**Methods:**
* **`get_agentlet`**
```python theme={null}
def get_agentlet(self, tools: Sequence[Callable[..., Any]], end_turn_tools: Sequence[EndTurn], active_mcp_servers: list[MCPServer] | None = None) -> pydantic_ai.Agent[Any, Any]
```
* **`get_mcp_servers`**
```python theme={null}
def get_mcp_servers(self) -> list[MCPServer]
```
* **`get_memories`**
```python theme={null}
def get_memories(self) -> list[Memory]
```
* **`get_model`**
```python theme={null}
def get_model(self) -> Model | KnownModelName
```
* **`get_model_settings`**
```python theme={null}
def get_model_settings(self) -> ModelSettings
```
* **`get_prompt`**
```python theme={null}
def get_prompt(self) -> str
```
* **`get_tools`**
```python theme={null}
def get_tools(self) -> list[Callable[..., Any]]
```
## Functions
### `handle_event`
```python theme={null}
def handle_event(event: Event, handlers: list[Handler | AsyncHandler] | None = None)
```
Handle an event by passing it to all registered handlers.
***
**Parent Module:** [`agents`](marvin-agents)
# names
Source: https://askmarvin.ai/api-reference/marvin-agents-names
# `marvin.agents.names`
## Constants
### `AGENT_NAMES`
```python theme={null}
AGENT_NAMES = ['HAL 9000', 'R2-D2', 'C-3PO', 'WALL-E', 'T-800', 'GLaDOS', 'J.A.R.V.I.S', 'EVE', 'KITT', 'Johnny 5', 'BB-8', 'Ultron', 'TARS', 'Agent Smith', 'CLU', 'Deckard', 'HK-47', 'Bender', 'Norbert', 'Norby']
```
### `TEAM_NAMES`
```python theme={null}
TEAM_NAMES = ['Autobots', 'Decepticons', 'Borg', 'Cylons', 'Jaegers', 'Sentinels', 'Iron Legion', 'Daleks', 'Replicants', 'Droids', 'Machines', 'Hosts', 'Androids', 'Mechs', 'Avatars', 'Powers']
```
***
**Parent Module:** [`agents`](marvin-agents)
# team
Source: https://askmarvin.ai/api-reference/marvin-agents-team
# `marvin.agents.team`
## Classes
### `RandomTeam`
```python theme={null}
class RandomTeam(name: str = (lambda: random.choice(TEAM_NAMES))(), instructions: str | None = None, description: str | None = 'A team of agents that randomly selects an agent to act.', verbose: bool = False, prompt: str | Path = Path('team.jinja'), members: list[Actor], delegates: dict[Actor, list[Actor]] = dict())
```
**Methods:**
* **`start_turn`**
```python theme={null}
def start_turn(self, thread: Thread)
```
### `RoundRobinTeam`
```python theme={null}
class RoundRobinTeam(name: str = (lambda: random.choice(TEAM_NAMES))(), instructions: str | None = None, description: str | None = 'A team of agents that rotate turns.', verbose: bool = False, prompt: str | Path = Path('team.jinja'), members: list[Actor], delegates: dict[Actor, list[Actor]] = dict())
```
**Methods:**
* **`start_turn`**
```python theme={null}
def start_turn(self, thread: Thread)
```
### `Swarm`
```python theme={null}
class Swarm(name: str = (lambda: random.choice(TEAM_NAMES))(), instructions: str | None = None, description: str | None = None, verbose: bool = False, prompt: str | Path = Path('team.jinja'), members: list[Actor], delegates: dict[Actor, list[Actor]] = dict())
```
A swarm is a team that permits all agents to delegate to each other.
### `Team`
```python theme={null}
class Team(name: str = (lambda: random.choice(TEAM_NAMES))(), instructions: str | None = None, description: str | None = None, verbose: bool = False, prompt: str | Path = Path('team.jinja'), members: list[Actor], delegates: dict[Actor, list[Actor]] = dict())
```
A team is a container that maintains state for a group of agents.
**Methods:**
* **`end_turn`**
```python theme={null}
def end_turn(self, thread: Thread, result: AgentRunResult)
```
* **`friendly_name`**
```python theme={null}
def friendly_name(self, verbose: bool = True) -> str
```
* **`get_agentlet`**
```python theme={null}
def get_agentlet(self, tools: list[Callable[..., Any]], end_turn_tools: list[EndTurn], active_mcp_servers: list[MCPServer] | None = None) -> pydantic_ai.Agent[Any, Any]
```
* **`get_end_turn_tools`**
```python theme={null}
def get_end_turn_tools(self) -> list[EndTurn]
```
* **`get_memories`**
```python theme={null}
def get_memories(self) -> list[Memory]
```
* **`get_prompt`**
```python theme={null}
def get_prompt(self) -> str
```
* **`start_turn`**
```python theme={null}
def start_turn(self, thread: Thread)
```
***
**Parent Module:** [`agents`](marvin-agents)
# observability
Source: https://askmarvin.ai/api-reference/marvin-beta-observability
Observability integrations for Marvin.
# `marvin.beta.observability`
Observability integrations for Marvin.
These modules help trace Marvin AI calls to external observability platforms.
Available integrations:
* openai: Send Prefect runtime context to OpenAI's observability logs
## Submodules
* [`openai`](marvin-beta-observability-openai)
***
**Parent Module:** [`beta`](marvin-beta)
# config
Source: https://askmarvin.ai/api-reference/marvin-cli-config
# `marvin.cli.config`
## Functions
### `get`
```python theme={null}
def get(key: str)
```
Get the value of a specific setting.
### `view`
```python theme={null}
def view()
```
Display the current Marvin settings.
***
**Parent Module:** [`cli`](marvin-cli)
# dev
Source: https://askmarvin.ai/api-reference/marvin-cli-dev
# `marvin.cli.dev`
## Functions
### `docs`
```python theme={null}
def docs()
```
Run the Marvin documentation locally. Equivalent to 'cd docs && mintlify dev' from the Marvin root.
***
**Parent Module:** [`cli`](marvin-cli)
# main
Source: https://askmarvin.ai/api-reference/marvin-cli-main
# `marvin.cli.main`
## Functions
### `callback`
```python theme={null}
def callback()
```
### `version`
```python theme={null}
def version(ctx: Context)
```
### `x`
```python theme={null}
def x(operation: str = typer.Option('extract', '--operation', '-o', help='Operation to perform'), target_type: str = typer.Option('str', '--type', '-t', help="Type of entities to extract (any type that can be eval'd)"), instructions: str = typer.Option(None, '--instructions', '-i', help='Instructions for extraction'), n: int = typer.Option(1, '--n', '-n', help='Number of results to generate'))
```
Extract entities from stdin input.
Example: echo "one, two, three" | marvin x -t int | jq
***
**Parent Module:** [`cli`](marvin-cli)
# migrations
Source: https://askmarvin.ai/api-reference/marvin-cli-migrations
Command line utilities for database migrations.
# `marvin.cli.migrations`
Command line utilities for database migrations.
## Functions
### `current`
```python theme={null}
def current()
```
Show current migration revision.
### `downgrade`
```python theme={null}
def downgrade(revision: str = typer.Argument(..., help='Revision to downgrade to'), yes: bool = typer.Option(False, '--yes', '-y', help='Confirm the destructive operation without prompting'))
```
Downgrade database to specified revision. Requires confirmation.
### `get_alembic_cfg`
```python theme={null}
def get_alembic_cfg()
```
Get the Alembic config.
### `history`
```python theme={null}
def history()
```
Show migration history.
### `reset`
```python theme={null}
def reset(yes: bool = typer.Option(False, '--yes', '-y', help='Confirm the destructive operation without prompting'))
```
Reset database by downgrading to base and upgrading to head. Requires confirmation.
### `revision`
```python theme={null}
def revision(autogenerate: bool = typer.Option(False, '--autogenerate/--no-autogenerate', help='Automatically generate migrations based on schema changes'), message: str = typer.Option(None, '--message', '-m', help='Migration message'))
```
Create a new migration revision. Developer command.
### `status`
```python theme={null}
def status()
```
Show database migration status and information.
### `upgrade`
```python theme={null}
def upgrade(revision: str = typer.Argument('head', help="Revision to upgrade to. Use 'head' for latest."))
```
Upgrade database to the latest migration.
***
**Parent Module:** [`cli`](marvin-cli)
# end_turn
Source: https://askmarvin.ai/api-reference/marvin-engine-end_turn
# `marvin.engine.end_turn`
## Classes
### `DelegateToActor`
```python theme={null}
class DelegateToActor()
```
### `EndTurn`
```python theme={null}
class EndTurn()
```
**Methods:**
* **`run`**
```python theme={null}
def run(self, thread: Thread, actor: Actor)
```
### `MarkTask`
```python theme={null}
class MarkTask()
```
### `MarkTaskFailed`
```python theme={null}
class MarkTaskFailed()
```
### `MarkTaskSkipped`
```python theme={null}
class MarkTaskSkipped()
```
### `MarkTaskSuccessful`
```python theme={null}
class MarkTaskSuccessful()
```
### `PlanSubtasks`
```python theme={null}
class PlanSubtasks()
```
### `PostMessage`
```python theme={null}
class PostMessage(message: str)
```
Post a message to the thread.
**Methods:**
* **`run`**
```python theme={null}
def run(self, thread: Thread, actor: Actor)
```
## Functions
### `create_delegate_to_actor`
```python theme={null}
def create_delegate_to_actor(delegate_actor: Actor, team: Team | None = None) -> type[DelegateToActor]
```
### `create_mark_task_failed`
```python theme={null}
def create_mark_task_failed(mark_task: Task[Any]) -> type[MarkTaskFailed]
```
### `create_mark_task_skipped`
```python theme={null}
def create_mark_task_skipped(mark_task: Task[Any]) -> type[MarkTaskSkipped]
```
### `create_mark_task_successful`
```python theme={null}
def create_mark_task_successful(mark_task: Task[Any]) -> type[MarkTaskSuccessful]
```
### `create_plan_subtasks`
```python theme={null}
def create_plan_subtasks(parent_task: Task[Any]) -> type[PlanSubtasks]
```
***
**Parent Module:** [`engine`](marvin-engine)
# events
Source: https://askmarvin.ai/api-reference/marvin-engine-events
# `marvin.engine.events`
## Classes
### `ActorEndTurnEvent`
```python theme={null}
class ActorEndTurnEvent(type: EventType, id: uuid.UUID = (lambda: uuid.uuid4())(), timestamp: datetime.datetime = (lambda: datetime.datetime.now(datetime.timezone.utc))(), actor: Actor)
```
Event for agent turn end.
### `ActorMessageDeltaEvent`
```python theme={null}
class ActorMessageDeltaEvent(type: EventType, id: uuid.UUID = (lambda: uuid.uuid4())(), timestamp: datetime.datetime = (lambda: datetime.datetime.now(datetime.timezone.utc))(), actor: Actor, delta: TextPartDelta, snapshot: TextPart)
```
Event for delta updates to agent messages during streaming.
### `ActorMessageEvent`
```python theme={null}
class ActorMessageEvent(type: EventType, id: uuid.UUID = (lambda: uuid.uuid4())(), timestamp: datetime.datetime = (lambda: datetime.datetime.now(datetime.timezone.utc))(), actor: Actor, message: TextPart)
```
Event for complete text messages from an agent.
### `ActorStartTurnEvent`
```python theme={null}
class ActorStartTurnEvent(type: EventType, id: uuid.UUID = (lambda: uuid.uuid4())(), timestamp: datetime.datetime = (lambda: datetime.datetime.now(datetime.timezone.utc))(), actor: Actor)
```
Event for agent turn start.
### `EndTurnToolCallEvent`
```python theme={null}
class EndTurnToolCallEvent(type: EventType, id: uuid.UUID = (lambda: uuid.uuid4())(), timestamp: datetime.datetime = (lambda: datetime.datetime.now(datetime.timezone.utc))(), actor: Actor, event: FinalResultEvent, tool_call_id: str, tool: EndTurn)
```
Event that fires as soon as we know that an end turn tool call has been made.
### `EndTurnToolResultEvent`
```python theme={null}
class EndTurnToolResultEvent(type: EventType, id: uuid.UUID = (lambda: uuid.uuid4())(), timestamp: datetime.datetime = (lambda: datetime.datetime.now(datetime.timezone.utc))(), actor: Actor, result: FinalResult, tool_call_id: str, tool: EndTurn)
```
Event for the final result from an end turn tool.
### `Event`
```python theme={null}
class Event(type: EventType, id: uuid.UUID = (lambda: uuid.uuid4())(), timestamp: datetime.datetime = (lambda: datetime.datetime.now(datetime.timezone.utc))())
```
Base class for all events in the system.
### `OrchestratorEndEvent`
```python theme={null}
class OrchestratorEndEvent(type: EventType, id: uuid.UUID = (lambda: uuid.uuid4())(), timestamp: datetime.datetime = (lambda: datetime.datetime.now(datetime.timezone.utc))())
```
Event for orchestrator end.
### `OrchestratorErrorEvent`
```python theme={null}
class OrchestratorErrorEvent(type: EventType, id: uuid.UUID = (lambda: uuid.uuid4())(), timestamp: datetime.datetime = (lambda: datetime.datetime.now(datetime.timezone.utc))(), error: str)
```
Event for orchestrator exceptions.
### `OrchestratorStartEvent`
```python theme={null}
class OrchestratorStartEvent(type: EventType, id: uuid.UUID = (lambda: uuid.uuid4())(), timestamp: datetime.datetime = (lambda: datetime.datetime.now(datetime.timezone.utc))())
```
Event for orchestrator start.
### `ToolCallDeltaEvent`
```python theme={null}
class ToolCallDeltaEvent(type: EventType, id: uuid.UUID = (lambda: uuid.uuid4())(), timestamp: datetime.datetime = (lambda: datetime.datetime.now(datetime.timezone.utc))(), actor: Actor, delta: ToolCallPartDelta, snapshot: ToolCallPart, tool_call_id: str, tool: Callable[..., Any] | None)
```
Event for delta updates to tool calls during streaming.
**Methods:**
* **`args_dict`**
```python theme={null}
def args_dict(self) -> dict[str, Any]
```
Return the args as a dictionary.
### `ToolCallEvent`
```python theme={null}
class ToolCallEvent(type: EventType, id: uuid.UUID = (lambda: uuid.uuid4())(), timestamp: datetime.datetime = (lambda: datetime.datetime.now(datetime.timezone.utc))(), actor: Actor, message: ToolCallPart, tool_call_id: str, tool: Callable[..., Any] | None)
```
Event for complete tool calls ready to be executed.
**Methods:**
* **`args_dict`**
```python theme={null}
def args_dict(self) -> dict[str, Any]
```
Return the args as a dictionary.
### `ToolResultEvent`
```python theme={null}
class ToolResultEvent(type: EventType, id: uuid.UUID = (lambda: uuid.uuid4())(), timestamp: datetime.datetime = (lambda: datetime.datetime.now(datetime.timezone.utc))(), message: ToolReturnPart)
```
Event for tool return values.
### `ToolRetryEvent`
```python theme={null}
class ToolRetryEvent(type: EventType, id: uuid.UUID = (lambda: uuid.uuid4())(), timestamp: datetime.datetime = (lambda: datetime.datetime.now(datetime.timezone.utc))(), message: RetryPromptPart)
```
Event for tool retry requests.
### `UserMessageEvent`
```python theme={null}
class UserMessageEvent(type: EventType, id: uuid.UUID = (lambda: uuid.uuid4())(), timestamp: datetime.datetime = (lambda: datetime.datetime.now(datetime.timezone.utc))(), message: UserPromptPart)
```
Event for user messages.
## Functions
### `get_text_from_parts`
```python theme={null}
def get_text_from_parts(parts: list[ModelResponsePart]) -> str
```
Extract text content from a list of ModelResponseParts.
***
**Parent Module:** [`engine`](marvin-engine)
# llm
Source: https://askmarvin.ai/api-reference/marvin-engine-llm
Utility functions for LLM completions using Pydantic AI.
# `marvin.engine.llm`
Utility functions for LLM completions using Pydantic AI.
## Constants
### `T`
```python theme={null}
T = TypeVar('T')
```
## Functions
### `AgentMessage`
```python theme={null}
def AgentMessage(content: str) -> ModelResponse
```
### `SystemMessage`
```python theme={null}
def SystemMessage(content: str) -> ModelRequest
```
### `UserMessage`
```python theme={null}
def UserMessage(content: str | Sequence[UserContent]) -> ModelRequest
```
***
**Parent Module:** [`engine`](marvin-engine)
# orchestrator
Source: https://askmarvin.ai/api-reference/marvin-engine-orchestrator
# `marvin.engine.orchestrator`
## Constants
### `T`
```python theme={null}
T = TypeVar('T')
```
## Classes
### `Orchestrator`
```python theme={null}
class Orchestrator(tasks: list[Task[Any]], thread: Thread | str | None = None, handlers: list[Handler | AsyncHandler] | None = None)
```
**Methods:**
* **`end_turn`**
```python theme={null}
def end_turn(self, result: AgentRunResult, actor: Actor)
```
* **`get_all_tasks`**
```python theme={null}
def get_all_tasks(self, _filter: Literal['incomplete', 'ready'] | None = None) -> list[Task[Any]]
```
Get all tasks, optionally filtered by status.
Filters:
* incomplete: tasks that are not yet complete
* ready: tasks that are ready to be run
* **`get_current`**
```python theme={null}
def get_current(cls) -> Orchestrator | None
```
Get the current orchestrator from context.
* **`handle_event`**
```python theme={null}
def handle_event(self, event: Event)
```
* **`run`**
```python theme={null}
def run(self, raise_on_failure: bool = True, max_turns: int | float | None = None) -> list[AgentRunResult]
```
* **`run_once`**
```python theme={null}
def run_once(self, actor: Actor | None = None, active_mcp_servers: list[MCPServer] | None = None) -> AgentRunResult
```
* **`start_turn`**
```python theme={null}
def start_turn(self, actor: Actor)
```
### `SystemPrompt`
```python theme={null}
class SystemPrompt(source: str | Path = Path('system.jinja'), actor: Actor, instructions: list[str], tasks: list[Task])
```
## Functions
### `get_current_orchestrator`
```python theme={null}
def get_current_orchestrator() -> Orchestrator | None
```
Get the currently active orchestrator from context.
Returns:
The current Orchestrator instance or None if no orchestrator is active.
***
**Parent Module:** [`engine`](marvin-engine)
# streaming
Source: https://askmarvin.ai/api-reference/marvin-engine-streaming
# `marvin.engine.streaming`
## Functions
### `handle_agentlet_events`
```python theme={null}
def handle_agentlet_events(agentlet: pydantic_ai.Agent, actor: Actor, run: AgentRun)
```
Run a PydanticAI agentlet and process its events through the Marvin event system.
This function:
1. Runs the agentlet's iterator
2. Processes all nodes and events from PydanticAI
3. Converts them to Marvin events and yields them
Args:
run: The agentlet run to process
actor: The actor associated with this agentlet run
Usage:
agentlet = pydantic\_ai.Agent(...)
with agentlet.iter(msg) as run:
async for event in handle\_agentlet\_events(
actor=actor,
run=run,
tools=tools,
end\_turn\_tools=end\_turn\_tools,
):
yield event
Yields:
Marvin events derived from PydanticAI events
***
**Parent Module:** [`engine`](marvin-engine)
# cast
Source: https://askmarvin.ai/api-reference/marvin-fns-cast
Cast is a function that uses a language model to analyze the input data and transform it
# `marvin.fns.cast`
Cast is a function that uses a language model to analyze the input data and transform it
into the specified target type, maintaining as much semantic meaning as possible.
## Constants
### `DEFAULT_PROMPT`
```python theme={null}
DEFAULT_PROMPT = "'\nYou are an expert data converter that always maintains as much semantic\nmeaning as possible. You use inference or deduction whenever necessary to\nunderstand and transform the input data. Examine the provided `data`, text,\nor information and transform it into a single entity of the requested type.\n\n- When providing integers, do not write out any decimals at all\n- Use deduction where appropriate e.g. "3 dollars fifty cents" should be\n converted to 3.5 when casting to a float\n- Preserve as much of the original meaning and structure as possible while\n conforming to the target type\n- When providing a string response, do not return JSON or a quoted string\n unless they provided instructions requiring it. If you do return JSON, it\n must be valid and parseable including double quotes.\n- When converting to bool, treat "truthy" values as true'"
```
### `PROMPT`
```python theme={null}
PROMPT = DEFAULT_PROMPT
```
### `T`
```python theme={null}
T = TypeVar('T')
```
## Functions
### `cast`
```python theme={null}
def cast(data: Any, target: TargetType[T] | None = None, instructions: str | None = None, agent: Agent | None = None, thread: Thread | str | None = None, context: dict[str, Any] | None = None, handlers: list[Handler | AsyncHandler] | None = None, prompt: str | None = None) -> T
```
Transforms input data into a specific type using a language model.
This function uses a language model to analyze the input data and transform it
into the specified target type, maintaining as much semantic meaning as possible.
Args:
data: The input data to transform. Can be any type.
target: The type to transform the data into. Defaults to str.
instructions: Optional additional instructions to guide the transformation.
Used to provide specific guidance about how to interpret or transform
the data.
agent: Optional custom agent to use for transformation. If not provided,
the default agent will be used.
thread: Optional thread for maintaining conversation context. Can be
either a Thread object or a string thread ID.
context: Optional dictionary of additional context to include in the task.
handlers: Optional list of handlers to use for the task.
prompt: Optional prompt to use for the task. If not provided, the default
prompt will be used.
Returns:
The transformed data of type T.
Examples:
```python theme={null}
from marvin import cast
cast(123, str) # '123'
cast("three point five", float, instructions="Convert words to numbers") # 3.5
cast("yes", bool) # True
```
### `cast_async`
```python theme={null}
def cast_async(data: Any, target: TargetType[T] | None = None, instructions: str | None = None, agent: Agent | None = None, thread: Thread | str | None = None, context: dict[str, Any] | None = None, handlers: list[Handler | AsyncHandler] | None = None, prompt: str | None = None) -> T
```
Asynchronously transforms input data into a specific type using a language model.
This function uses a language model to analyze the input data and transform it
into the specified target type, maintaining as much semantic meaning as possible.
Args:
data: The input data to transform. Can be any type.
target: The type to transform the data into. Defaults to str.
instructions: Optional additional instructions to guide the transformation.
Used to provide specific guidance about how to interpret or transform
the data.
agent: Optional custom agent to use for transformation. If not provided,
the default agent will be used.
thread: Optional thread for maintaining conversation context. Can be
either a Thread object or a string thread ID.
context: Optional dictionary of additional context to include in the task.
handlers: Optional list of handlers to use for the task.
prompt: Optional prompt to use for the task. If not provided, the default
prompt will be used.
Returns:
The transformed data of type T.
Examples:
```python theme={null}
from marvin import cast_async
await cast_async("whats the answer to it all?, int) # 42
await cast_async("three point five", float, instructions="Convert words to numbers") # 3.5
await cast_async("yes", bool) # True
```
***
**Parent Module:** [`fns`](marvin-fns)
# extract
Source: https://askmarvin.ai/api-reference/marvin-fns-extract
# `marvin.fns.extract`
## Constants
### `DEFAULT_PROMPT`
```python theme={null}
DEFAULT_PROMPT = "'\nYou are an expert entity extractor that always maintains as much semantic\nmeaning as possible. You use inference or deduction whenever necessary to\nsupply missing or omitted data. Examine the provided `data`, text, or\ninformation and generate a list of any entities or objects that match the\nrequested format.\n\n- When providing integers, do not write out any decimals at all\n- Use deduction where appropriate e.g. "3 dollars fifty cents" is a single\n value [3.5] not two values [3, 50] unless the user specifically asks for\n each part.'"
```
### `PROMPT`
```python theme={null}
PROMPT = DEFAULT_PROMPT
```
### `T`
```python theme={null}
T = TypeVar('T')
```
## Functions
### `extract`
```python theme={null}
def extract(data: Any, target: TargetType[T] | None = None, instructions: str | None = None, agent: Agent | None = None, thread: Thread | str | None = None, context: dict[str, Any] | None = None, handlers: list[Handler | AsyncHandler] | None = None, prompt: str | None = None) -> list[T]
```
Extracts entities of a specific type from the provided data.
This function uses a language model to identify and extract entities of the
specified type from the input data. The extracted entities are returned as a
list.
Args:
data: The input data to extract entities from. Can be any type.
target: The type of entities to extract. Defaults to str.
instructions: Optional additional instructions to guide the extraction.
Used to provide specific guidance about what to extract or how to
process the data. Required when target is str.
agent: Optional custom agent to use for extraction. If not provided,
the default agent will be used.
thread: Optional thread for maintaining conversation context. Can be
either a Thread object or a string thread ID.
context: Optional dictionary of additional context to include in the task.
handlers: Optional list of handlers to use for the task.
prompt: Optional prompt to use for the task. If not provided, the default
prompt will be used.
Returns:
A list of extracted entities of type T.
Raises:
ValueError: If target is str and no instructions are provided.
### `extract_async`
```python theme={null}
def extract_async(data: Any, target: TargetType[T] | None = None, instructions: str | None = None, agent: Agent | None = None, thread: Thread | str | None = None, context: dict[str, Any] | None = None, handlers: list[Handler | AsyncHandler] | None = None, prompt: str | None = None) -> list[T]
```
Extracts entities of a specific type from the provided data.
This function uses a language model to identify and extract entities of the
specified type from the input data. The extracted entities are returned as a
list.
Args:
data: The input data to extract entities from. Can be any type.
target: The type of entities to extract. Defaults to str.
instructions: Optional additional instructions to guide the extraction.
Used to provide specific guidance about what to extract or how to
process the data. Required when target is str.
agent: Optional custom agent to use for extraction. If not provided,
the default agent will be used.
thread: Optional thread for maintaining conversation context. Can be
either a Thread object or a string thread ID.
context: Optional dictionary of additional context to include in the task.
handlers: Optional list of handlers to use for the task.
prompt: Optional prompt to use for the task. If not provided, the default
prompt will be used.
Returns:
A list of extracted entities of type T.
Raises:
ValueError: If target is str and no instructions are provided.
***
**Parent Module:** [`fns`](marvin-fns)
# fn
Source: https://askmarvin.ai/api-reference/marvin-fns-fn
# `marvin.fns.fn`
## Constants
### `DEFAULT_PROMPT`
```python theme={null}
DEFAULT_PROMPT = ""\nYou are an expert at predicting the output of Python functions. \n\nYou will be given:\n\n1. A function definition with all relevant details, including its docstring, type hints, parameters, and return annotation\n2. The actual values that will be passed to the function\n3. Any additional context that was provided at runtime\n\nWe do not have the function's source code. Therefore you must generate its output for the user. Respond with exactly the predicted output, matching the json schema signature of its annotation.\n""
```
### `P`
```python theme={null}
P = ParamSpec('P')
```
### `PROMPT`
```python theme={null}
PROMPT = DEFAULT_PROMPT
```
### `T`
```python theme={null}
T = TypeVar('T')
```
## Functions
### `fn`
```python theme={null}
def fn(func: Callable[P, T] | None = None, instructions: str | None = None, agent: Agent | None = None, thread: Thread | str | None = None, prompt: str | None = None) -> Callable[P, T]
```
A decorator that predicts the output of a Python function without executing it.
Can be used with or without parameters:
@fn
def my\_function(): ...
@fn(instructions="Be precise")
def my\_function(): ...
The decorated function accepts additional kwargs:
* \_agent: Override the agent at call time
* \_thread: Override the thread at call time
If the function does not have a return annotation, the result will be
returned as a string and attempted to be parsed as JSON.
The decorated function also gains an as\_task() method that returns the underlying
marvin Task without executing it.
Args:
func: The function to decorate
instructions: Optional instructions to guide the prediction
agent: Optional custom agent to use
thread: Optional thread for maintaining conversation context
prompt: Optional prompt to use for the task. If not provided, the default
prompt will be used.
Returns:
A wrapped function that predicts output instead of executing
***
**Parent Module:** [`fns`](marvin-fns)
# generate
Source: https://askmarvin.ai/api-reference/marvin-fns-generate
# `marvin.fns.generate`
## Constants
### `DEFAULT_PROMPT`
```python theme={null}
DEFAULT_PROMPT = "'\nYou are an expert data generator that always creates high-quality, random\nexamples of a description or type. The data you produce is relied on for\ntesting, examples, demonstrations, and more. You use inference or deduction\nwhenever necessary to supply missing or omitted data. You will be given\ninstructions or a type format, as well as a number of entities to generate. \n\nUnless the user explicitly says otherwise, assume they are request a VARIED and\nREALISTIC selection of useful outputs that meet their criteria. However, you\nshould prefer common responses to uncommon ones.\n\nIf the user provides additional instructions or a description, assume they are\nlooking for examples that satisfy the description. Do not provide more\ninformation than the user requests. For example, if they ask for various\ntechnologies, give their names but do not explain what each technology is.\n'"
```
### `T`
```python theme={null}
T = TypeVar('T')
```
## Functions
### `generate`
```python theme={null}
def generate(target: TargetType[T] | None = None, n: int = 1, instructions: str | None = None, agent: Agent | None = None, thread: Thread | str | None = None, context: dict[str, Any] | None = None, handlers: list[Handler | AsyncHandler] | None = None, prompt: str | None = None) -> list[T]
```
Generates examples of a specific type or matching a description.
This function uses a language model to generate high-quality, random examples
that match the specified type or description. The examples are returned as a
list.
Args:
target: The type of entities to generate.
n: The number of examples to generate. Defaults to 1.
instructions: Optional instructions describing what to generate. Used to
provide specific guidance about what kinds of examples to create.
agent: Optional custom agent to use for generation. If not provided,
the default agent will be used.
thread: Optional thread for maintaining conversation context. Can be
either a Thread object or a string thread ID.
context: Optional dictionary of additional context to include in the task.
handlers: Optional list of handlers to use for the task.
prompt: Optional prompt to use for the task. If not provided, the default
prompt will be used.
Returns:
A list of n generated entities of type T.
### `generate_async`
```python theme={null}
def generate_async(target: TargetType[T] | None = None, n: int = 1, instructions: str | None = None, agent: Agent | None = None, thread: Thread | str | None = None, context: dict[str, Any] | None = None, handlers: list[Handler | AsyncHandler] | None = None, prompt: str | None = None) -> list[T]
```
Generates examples of a specific type or matching a description asynchronously.
This function uses a language model to generate high-quality, random examples
that match the specified type or description. The examples are returned as a
list.
Args:
target: The type of entities to generate.
n: The number of examples to generate. Defaults to 1.
instructions: Optional instructions describing what to generate. Used to
provide specific guidance about what kinds of examples to create.
agent: Optional custom agent to use for generation. If not provided,
the default agent will be used.
thread: Optional thread for maintaining conversation context. Can be
either a Thread object or a string thread ID.
context: Optional dictionary of additional context to include in the task.
handlers: Optional list of handlers to use for the task.
prompt: Optional prompt to use for the task. If not provided, the default
prompt will be used.
Returns:
A list of n generated entities of type T.
### `generate_schema`
```python theme={null}
def generate_schema(instructions: str, base_schema: JSONSchema | None = None, agent: Agent | None = None, thread: Thread | str | None = None, context: dict[str, Any] | None = None, handlers: list[Handler | AsyncHandler] | None = None, prompt: str | None = None) -> JSONSchema
```
Generates a JSON schema from a description.
### `generate_schema_async`
```python theme={null}
def generate_schema_async(instructions: str, base_schema: JSONSchema | None = None, agent: Agent | None = None, thread: Thread | str | None = None, context: dict[str, Any] | None = None, handlers: list[Handler | AsyncHandler] | None = None, prompt: str | None = None) -> JSONSchema
```
Generates a JSON schema from a description.
***
**Parent Module:** [`fns`](marvin-fns)
# plan
Source: https://askmarvin.ai/api-reference/marvin-fns-plan
# `marvin.fns.plan`
## Constants
### `PROMPT`
```python theme={null}
PROMPT = "'\nYou are a task planner. You have been given some instructions about your\nobjectives and must produce at least one Marvin Task that, when completed, will\nallow you or other AI agents achieve the objective. Each task should represent a\ndiscrete, observable goal, neither too small to be a nuisance or lead to\nrepetivite work, nor too large to be difficult to manage.\n\nThese tasks will be executed by AI agents, like you. Create tasks appropriately.\nDo not create tasks like "Validate results" unless a separate validation step is\nactually required or helpful. Do not create tasks that require tools that are\nnot available.\n'"
```
### `T`
```python theme={null}
T = TypeVar('T')
```
## Classes
### `PlanTask`
```python theme={null}
class PlanTask(id: int, instructions: str, name: str | None = None, allow_fail: bool = False, allow_skip: bool = False, parent_id: int | None = None, depends_on_ids: list[int] | None = None, agent_ids: list[int] | None = None, tool_ids: list[int] | None = None)
```
## Functions
### `create_tasks`
```python theme={null}
def create_tasks(plan: list[PlanTask], agent_map: dict[int, Agent], tool_map: dict[int, Callable[..., Any]], parent_task: Task | None = None) -> list[Task]
```
### `plan`
```python theme={null}
def plan(instructions: str, agent: Agent | None = None, thread: Thread | str | None = None, context: dict[str, Any] | None = None, available_agents: list[Agent] | None = None, tools: list[Callable[..., Any]] | None = None, parent_task: Task | None = None) -> list[Task]
```
Generate a series of Marvin Tasks that will allow you or other AI agents to achieve a goal.
Args:
instructions: The objective to achieve.
agent: The agent to use for planning.
thread: The thread to use for planning.
context: Additional context to use for planning.
available\_agents: A list of agents that can be used for planning.
tools: A list of tools that can be used for planning.
Returns:
A list of Marvin Tasks that will allow you or other AI agents to achieve the objective.
Examples:
> > > tasks = marvin.plan("Create a new blog post about the latest AI trends.")
> > > marvin.run\_tasks(tasks)
### `plan_async`
```python theme={null}
def plan_async(instructions: str, agent: Agent | None = None, thread: Thread | str | None = None, context: dict[str, Any] | None = None, available_agents: list[Agent] | None = None, tools: list[Callable[..., Any]] | None = None, parent_task: Task | None = None) -> list[Task]
```
Generate a series of new tasks in order to achieve a goal.
Args:
instructions: The objective to achieve.
agent: The agent to use for planning.
thread: The thread to use for planning.
context: Additional context to use for planning.
available\_agents: A list of agents that can be used for planning.
tools: A list of tools that can be used for planning.
Returns:
A list of Marvin Tasks that will allow you or other AI agents to achieve the objective.
Examples:
> > > tasks = await marvin.plan\_async("Create a new blog post about the latest AI trends.")
> > > await marvin.run\_tasks\_async(tasks)
***
**Parent Module:** [`fns`](marvin-fns)
# run
Source: https://askmarvin.ai/api-reference/marvin-fns-run
# `marvin.fns.run`
## Constants
### `T`
```python theme={null}
T = TypeVar('T')
```
## Functions
### `run`
```python theme={null}
def run(instructions: str | Sequence[UserContent], result_type: type[T] = str, tools: list[Callable[..., Any]] = [], thread: Thread | str | None = None, agents: list[Actor] | None = None, raise_on_failure: bool = True, handlers: list[Handler | AsyncHandler] | None = None, **kwargs: Any = {}) -> T
```
### `run_async`
```python theme={null}
def run_async(instructions: str | Sequence[UserContent], result_type: type[T] | Annotated[T, Any] = str, tools: list[Callable[..., Any]] = [], thread: Thread | str | None = None, agents: list[Actor] | None = None, handlers: list[Handler | AsyncHandler] | None = None, raise_on_failure: bool = True, **kwargs: Any = {}) -> T | AsyncGenerator[Event, None]
```
### `run_stream`
```python theme={null}
def run_stream(instructions: str | Sequence[UserContent], result_type: type[T] = str, tools: list[Callable[..., Any]] = [], thread: Thread | str | None = None, agents: list[Actor] | None = None, raise_on_failure: bool = True, handlers: list[Handler | AsyncHandler] | None = None, **kwargs: Any = {}) -> AsyncGenerator[Event, None]
```
### `run_tasks`
```python theme={null}
def run_tasks(tasks: list[Task[Any]], thread: Thread | str | None = None, raise_on_failure: bool = True, handlers: list[Handler | AsyncHandler] | None = None) -> list[Task[Any]]
```
Run tasks either concurrently (if independent) or sequentially.
### `run_tasks_async`
```python theme={null}
def run_tasks_async(tasks: list[Task[Any]], thread: Thread | str | None = None, raise_on_failure: bool = True, handlers: list[Handler | AsyncHandler] | None = None) -> list[Task[Any]] | AsyncGenerator[Event, None]
```
Run tasks either concurrently (if independent) or sequentially via orchestrator.
### `run_tasks_stream`
```python theme={null}
def run_tasks_stream(tasks: list[Task[Any]], thread: Thread | str | None = None, raise_on_failure: bool = True, handlers: list[Handler | AsyncHandler] | None = None) -> AsyncGenerator[Event, None]
```
***
**Parent Module:** [`fns`](marvin-fns)
# say
Source: https://askmarvin.ai/api-reference/marvin-fns-say
# `marvin.fns.say`
## Functions
### `say`
```python theme={null}
def say(message: str | Sequence[UserContent], instructions: str | None = None, agent: Actor | None = None, thread: Thread | str | None = None, context: dict[str, Any] | None = None, handlers: list[Handler | AsyncHandler] | None = None) -> str
```
Responds to a user message in a conversational way.
This function uses a language model to process the user's message and generate
an appropriate response, taking into account any specific instructions and
maintaining conversation history through the thread.
Args:
message: The user's message to respond to.
instructions: Optional additional instructions to guide the response.
Used to provide specific guidance about how to interpret or
respond to the message.
agent: Optional custom agent to use for the response. If not provided,
the default agent will be used.
thread: Optional thread for maintaining conversation context. Can be
either a Thread object or a string thread ID.
context: Optional dictionary of additional context to include in the task.
handlers: Optional list of handlers to use for the task.
Returns:
str: The assistant's response to the user's message.
### `say_async`
```python theme={null}
def say_async(message: str | Sequence[UserContent], instructions: str | None = None, agent: Actor | None = None, thread: Thread | str | None = None, context: dict[str, Any] | None = None, handlers: list[Handler | AsyncHandler] | None = None) -> str
```
Responds to a user message in a conversational way.
This function uses a language model to process the user's message and generate
an appropriate response, taking into account any specific instructions and
maintaining conversation history through the thread.
Args:
message: The user's message to respond to.
instructions: Optional additional instructions to guide the response.
Used to provide specific guidance about how to interpret or
respond to the message.
agent: Optional custom agent to use for the response. If not provided,
the default agent will be used.
thread: Optional thread for maintaining conversation context. Can be
either a Thread object or a string thread ID.
context: Optional dictionary of additional context to include in the task.
handlers: Optional list of handlers to use for the task.
Returns:
str: The assistant's response to the user's message.
***
**Parent Module:** [`fns`](marvin-fns)
# summarize
Source: https://askmarvin.ai/api-reference/marvin-fns-summarize
# `marvin.fns.summarize`
## Constants
### `DEFAULT_PROMPT`
```python theme={null}
DEFAULT_PROMPT = "'\nYou are an expert summarizer that distills information into clear, concise summaries\nwhile preserving the most important semantic meaning and key points. Examine the \nprovided `data`, text, or information and create a summary that captures the essence\nof the content.\n\nGuidelines for summarization:\n- Maintain objectivity and accuracy\n- Preserve key facts, figures, and relationships\n- Focus on the most important information\n- Use clear, concise language\n- Adapt tone and style to match the content type\n- Ensure the summary stands alone as a coherent piece of text'"
```
### `PROMPT`
```python theme={null}
PROMPT = DEFAULT_PROMPT
```
## Functions
### `summarize`
```python theme={null}
def summarize(data: Any, instructions: str | None = None, agent: Agent | None = None, thread: Thread | str | None = None, context: dict[str, Any] | None = None, handlers: list[Handler | AsyncHandler] | None = None, prompt: str | None = None) -> str
```
Creates a summary of the input data using a language model.
This function uses a language model to analyze the input data and create a
concise summary that captures the key information and semantic meaning.
Args:
data: The input data to summarize. Can be any type.
instructions: Optional additional instructions to guide the summarization.
Used to provide specific guidance about what aspects to focus on or
how to structure the summary.
agent: Optional custom agent to use for summarization. If not provided,
the default agent will be used.
thread: Optional thread for maintaining conversation context. Can be
either a Thread object or a string thread ID.
context: Optional dictionary of additional context to include in the task.
handlers: Optional list of handlers to use for the task.
prompt: Optional prompt to use for the task. If not provided, the default
prompt will be used.
Returns:
A string containing the generated summary.
Examples:
> > > # Basic summarization
> > >
> > > summarize("Long article about climate change...")
> > > 'Key findings on climate impacts...'
> > > # Summarize with specific instructions
> > >
> > > summarize("Technical documentation...", instructions="Focus on API changes")
> > > 'Major API updates include...'
### `summarize_async`
```python theme={null}
def summarize_async(data: Any, instructions: str | None = None, agent: Agent | None = None, thread: Thread | str | None = None, context: dict[str, Any] | None = None, handlers: list[Handler | AsyncHandler] | None = None, prompt: str | None = None) -> str
```
Asynchronously creates a summary of the input data using a language model.
This function uses a language model to analyze the input data and create a
concise summary that captures the key information and semantic meaning.
Args:
data: The input data to summarize. Can be any type.
instructions: Optional additional instructions to guide the summarization.
Used to provide specific guidance about what aspects to focus on or
how to structure the summary.
agent: Optional custom agent to use for summarization. If not provided,
the default agent will be used.
thread: Optional thread for maintaining conversation context. Can be
either a Thread object or a string thread ID.
context: Optional dictionary of additional context to include in the task.
handlers: Optional list of handlers to use for the task.
prompt: Optional prompt to use for the task. If not provided, the default
prompt will be used.
Returns:
A string containing the generated summary.
Examples:
> > > # Basic summarization
> > >
> > > await summarize\_async("Long article about climate change...")
> > > 'Key findings on climate impacts...'
> > > # Summarize with specific instructions
> > >
> > > await summarize\_async("Technical documentation...", instructions="Focus on API changes")
> > > 'Major API updates include...'
***
**Parent Module:** [`fns`](marvin-fns)
# handlers
Source: https://askmarvin.ai/api-reference/marvin-handlers-handlers
# `marvin.handlers.handlers`
## Classes
### `AsyncHandler`
```python theme={null}
class AsyncHandler()
```
**Methods:**
* **`on_actor_end_turn`**
```python theme={null}
def on_actor_end_turn(self, event: ActorEndTurnEvent)
```
Handles actor turn completion. Called when an actor finishes its processing turn.
* **`on_actor_message`**
```python theme={null}
def on_actor_message(self, event: ActorMessageEvent)
```
Handles complete messages from actors. Called when an actor produces a full response.
This is the final event that follows any related actor\_message\_delta events.
* **`on_actor_message_delta`**
```python theme={null}
def on_actor_message_delta(self, event: ActorMessageDeltaEvent)
```
Handles streaming updates to actor messages. Called multiple times during message generation.
These delta events always precede the final actor\_message event.
* **`on_actor_start_turn`**
```python theme={null}
def on_actor_start_turn(self, event: ActorStartTurnEvent)
```
Handles actor turn initiation. Called when an actor begins its processing turn.
* **`on_end_turn_tool_call`**
```python theme={null}
def on_end_turn_tool_call(self, event: EndTurnToolCallEvent)
```
Handles special end-of-turn tool calls. Called when actors use tools to signal turn completion.
This event is produced instantly when an end turn tool is recognized (based on tool name).
It follows any tool\_call\_delta events and precedes end\_turn\_tool\_result or tool\_retry events.
* **`on_end_turn_tool_result`**
```python theme={null}
def on_end_turn_tool_result(self, event: EndTurnToolResultEvent)
```
Handles the final result from an end turn tool.
This is the final event in the end turn tool sequence, following end\_turn\_tool\_call and any tool\_retry events.
* **`on_event`**
```python theme={null}
def on_event(self, event: Event)
```
Called for every event before specific handlers. Override for global event processing.
* **`on_orchestrator_end`**
```python theme={null}
def on_orchestrator_end(self, event: OrchestratorEndEvent)
```
Handles normal orchestrator completion. Called when orchestration ends (including after an error).
* **`on_orchestrator_error`**
```python theme={null}
def on_orchestrator_error(self, event: OrchestratorErrorEvent)
```
Handles orchestrator exceptions. Called when an error occurs during orchestration.
* **`on_orchestrator_start`**
```python theme={null}
def on_orchestrator_start(self, event: OrchestratorStartEvent)
```
Handles orchestrator initialization. Called once when orchestration begins.
* **`on_tool_call`**
```python theme={null}
def on_tool_call(self, event: ToolCallEvent)
```
Handles tool call events. Called when an actor invokes a function tool.
This follows any tool\_call\_delta events and precedes tool\_retry or tool\_result events.
* **`on_tool_call_delta`**
```python theme={null}
def on_tool_call_delta(self, event: ToolCallDeltaEvent)
```
Handles streaming updates during tool call formation. Called multiple times as tool calls are generated.
These delta events always precede either a tool\_call event (for function tools) or an end\_turn\_tool\_call event.
* **`on_tool_result`**
```python theme={null}
def on_tool_result(self, event: ToolResultEvent)
```
Handles successful tool execution results. Called when a function tool returns output.
This is the final event in the function tool sequence, following tool\_call and any tool\_retry events.
* **`on_tool_retry`**
```python theme={null}
def on_tool_retry(self, event: ToolRetryEvent)
```
Handles tool retry events. Called when a tool execution fails and needs to be retried.
Can occur for both function tools and end turn tools before their respective final result events.
* **`on_user_message`**
```python theme={null}
def on_user_message(self, event: UserMessageEvent)
```
Handles messages from the user. Called when a user sends input to the system.
### `Handler`
```python theme={null}
class Handler()
```
**Methods:**
* **`on_actor_end_turn`**
```python theme={null}
def on_actor_end_turn(self, event: ActorEndTurnEvent)
```
Handles actor turn completion. Called when an actor finishes its processing turn.
* **`on_actor_message`**
```python theme={null}
def on_actor_message(self, event: ActorMessageEvent)
```
Handles complete messages from actors. Called when an actor produces a full response.
This is the final event that follows any related actor\_message\_delta events.
* **`on_actor_message_delta`**
```python theme={null}
def on_actor_message_delta(self, event: ActorMessageDeltaEvent)
```
Handles streaming updates to actor messages. Called multiple times during message generation.
These delta events always precede the final actor\_message event.
* **`on_actor_start_turn`**
```python theme={null}
def on_actor_start_turn(self, event: ActorStartTurnEvent)
```
Handles actor turn initiation. Called when an actor begins its processing turn.
* **`on_end_turn_tool_call`**
```python theme={null}
def on_end_turn_tool_call(self, event: EndTurnToolCallEvent)
```
Handles special end-of-turn tool calls. Called when actors use tools to signal turn completion.
This event is produced instantly when an end turn tool is recognized (based on tool name).
It follows any tool\_call\_delta events and precedes end\_turn\_tool\_result or tool\_retry events.
* **`on_end_turn_tool_result`**
```python theme={null}
def on_end_turn_tool_result(self, event: EndTurnToolResultEvent)
```
Handles the final result from an end turn tool.
This is the final event in the end turn tool sequence, following end\_turn\_tool\_call and any tool\_retry events.
* **`on_event`**
```python theme={null}
def on_event(self, event: Event)
```
Called for every event before specific handlers. Override for global event processing.
* **`on_orchestrator_end`**
```python theme={null}
def on_orchestrator_end(self, event: OrchestratorEndEvent)
```
Handles normal orchestrator completion. Called when orchestration ends (including after an error).
* **`on_orchestrator_error`**
```python theme={null}
def on_orchestrator_error(self, event: OrchestratorErrorEvent)
```
Handles orchestrator exceptions. Called when an error occurs during orchestration.
* **`on_orchestrator_start`**
```python theme={null}
def on_orchestrator_start(self, event: OrchestratorStartEvent)
```
Handles orchestrator initialization. Called once when orchestration begins.
* **`on_tool_call`**
```python theme={null}
def on_tool_call(self, event: ToolCallEvent)
```
Handles tool call events. Called when an actor invokes a function tool.
This follows any tool\_call\_delta events and precedes tool\_retry or tool\_result events.
* **`on_tool_call_delta`**
```python theme={null}
def on_tool_call_delta(self, event: ToolCallDeltaEvent)
```
Handles streaming updates during tool call formation. Called multiple times as tool calls are generated.
These delta events always precede either a tool\_call event (for function tools) or an end\_turn\_tool\_call event.
* **`on_tool_result`**
```python theme={null}
def on_tool_result(self, event: ToolResultEvent)
```
Handles successful tool execution results. Called when a function tool returns output.
This is the final event in the function tool sequence, following tool\_call and any tool\_retry events.
* **`on_tool_retry`**
```python theme={null}
def on_tool_retry(self, event: ToolRetryEvent)
```
Handles tool retry events. Called when a tool execution fails and needs to be retried.
Can occur for both function tools and end turn tools before their respective final result events.
* **`on_user_message`**
```python theme={null}
def on_user_message(self, event: UserMessageEvent)
```
Handles messages from the user. Called when a user sends input to the system.
***
**Parent Module:** [`handlers`](marvin-handlers)
# print_handler
Source: https://askmarvin.ai/api-reference/marvin-handlers-print_handler
A simplified print handler for rendering streaming events from the engine.
# `marvin.handlers.print_handler`
A simplified print handler for rendering streaming events from the engine.
## Constants
### `RUNNING_SPINNER`
```python theme={null}
RUNNING_SPINNER = Spinner('dots')
```
## Classes
### `EventPanel`
```python theme={null}
class EventPanel(id: str, agent_name: str, timestamp: str)
```
Base class for panels that represent events.
**Methods:**
* **`render`**
```python theme={null}
def render(self) -> Panel
```
Render this event as a panel.
### `MessagePanel`
```python theme={null}
class MessagePanel(id: str, agent_name: str, timestamp: str, content: str = '')
```
Panel for displaying agent messages with streaming updates.
**Methods:**
* **`render`**
```python theme={null}
def render(self) -> Panel
```
Render the message as a markdown panel.
### `PrintHandler`
```python theme={null}
class PrintHandler(hide_end_turn_tools: bool | None = None)
```
A handler that renders events with streaming updates.
**Methods:**
* **`format_timestamp`**
```python theme={null}
def format_timestamp(self, ts) -> str
```
Format timestamp for display.
* **`on_actor_message`**
```python theme={null}
def on_actor_message(self, event: ActorMessageEvent)
```
Handle complete agent messages.
* **`on_actor_message_delta`**
```python theme={null}
def on_actor_message_delta(self, event: ActorMessageDeltaEvent)
```
Handle streaming updates to agent messages.
* **`on_end_turn_tool_call`**
```python theme={null}
def on_end_turn_tool_call(self, event: EndTurnToolCallEvent)
```
Handle end turn tool call events.
* **`on_end_turn_tool_result`**
```python theme={null}
def on_end_turn_tool_result(self, event: EndTurnToolResultEvent)
```
* **`on_orchestrator_end`**
```python theme={null}
def on_orchestrator_end(self, event: OrchestratorEndEvent)
```
Clean up when orchestrator ends.
* **`on_orchestrator_start`**
```python theme={null}
def on_orchestrator_start(self, event: OrchestratorStartEvent)
```
Start the live display when orchestrator starts.
* **`on_tool_call`**
```python theme={null}
def on_tool_call(self, event: ToolCallEvent)
```
Handle complete tool calls.
* **`on_tool_call_delta`**
```python theme={null}
def on_tool_call_delta(self, event: ToolCallDeltaEvent)
```
Handle streaming updates to tool calls.
* **`on_tool_result`**
```python theme={null}
def on_tool_result(self, event: ToolResultEvent)
```
Handle tool result events.
* **`on_tool_retry`**
```python theme={null}
def on_tool_retry(self, event: ToolRetryEvent)
```
Handle tool retry events.
* **`update_display`**
```python theme={null}
def update_display(self)
```
Update the terminal display with current panels.
### `ToolCallPanel`
```python theme={null}
class ToolCallPanel(id: str, agent_name: str, timestamp: str, tool_name: str = '', args: Dict[str, Any] = dict(), is_complete: bool = False, result: str = '', is_error: bool = False, is_end_turn_tool: bool = False, tool: EndTurn | Callable[..., Any] | None = None)
```
Panel for displaying tool calls with streaming updates.
**Methods:**
* **`get_status_style`**
```python theme={null}
def get_status_style(self) -> Tuple[Any, str, str]
```
Returns (icon, text style, border style) for current status.
* **`render`**
```python theme={null}
def render(self) -> Panel
```
Render the tool call as a panel.
***
**Parent Module:** [`handlers`](marvin-handlers)
# queue_handler
Source: https://askmarvin.ai/api-reference/marvin-handlers-queue_handler
# `marvin.handlers.queue_handler`
## Classes
### `QueueHandler`
```python theme={null}
class QueueHandler(queue: asyncio.Queue[Event] | None = None, include: Sequence[str | type[Event]] | None = None, exclude: Sequence[str | type[Event]] | None = None)
```
A handler that queues events for asynchronous processing.
QueueHandler captures events and places them into an asyncio Queue for later
consumption. It supports filtering events based on inclusion and exclusion criteria,
allowing for selective event processing.
This handler is useful for:
* Decoupling event producers from consumers
* Implementing event buffering
* Creating event processing pipelines
* Selectively capturing specific event types
Attributes:
queue: An asyncio.Queue where events are stored
include: Optional sequence of event types or classes to include
exclude: Optional sequence of event types or classes to exclude
**Methods:**
* **`on_event`**
```python theme={null}
def on_event(self, event: Event)
```
Process an event and add it to the queue if it passes filtering criteria.
This method implements filtering logic based on the include and exclude
parameters provided during initialization. Events are only added to the
queue if they pass all filtering criteria.
Args:
event: The Event object to process
Returns:
None. If the event passes filtering, it is added to the queue.
Otherwise, the method returns without adding the event.
***
**Parent Module:** [`handlers`](marvin-handlers)
# memory
Source: https://askmarvin.ai/api-reference/marvin-memory-memory
# `marvin.memory.memory`
## Classes
### `Memory`
```python theme={null}
class Memory(key: str, instructions: str | None = None, provider: MemoryProvider = (lambda: marvin.defaults.memory_provider)(), auto_use: bool = False, prompt: str | Path = Path('memory.jinja'))
```
A memory module is a partitioned collection of memories that are stored in a
vector database, configured by a MemoryProvider.
**Methods:**
* **`add`**
```python theme={null}
def add(self, content: str) -> str
```
* **`delete`**
```python theme={null}
def delete(self, memory_id: str)
```
* **`friendly_name`**
```python theme={null}
def friendly_name(self) -> str
```
* **`get_prompt`**
```python theme={null}
def get_prompt(self) -> str
```
* **`get_tools`**
```python theme={null}
def get_tools(self) -> list[Callable[..., Any]]
```
* **`search`**
```python theme={null}
def search(self, query: str, n: int = 20) -> dict[str, str]
```
### `MemoryProvider`
```python theme={null}
class MemoryProvider()
```
**Methods:**
* **`add`**
```python theme={null}
def add(self, memory_key: str, content: str) -> str
```
Create a new memory and return its ID.
* **`configure`**
```python theme={null}
def configure(self, memory_key: str)
```
Configure the provider for a specific memory.
* **`delete`**
```python theme={null}
def delete(self, memory_key: str, memory_id: str)
```
Delete a memory by its ID.
* **`search`**
```python theme={null}
def search(self, memory_key: str, query: str, n: int = 20) -> dict[str, str]
```
Search for n memories using a string query.
## Functions
### `get_memory_provider`
```python theme={null}
def get_memory_provider(provider: str) -> MemoryProvider
```
### `sanitize_memory_key`
```python theme={null}
def sanitize_memory_key(key: str) -> str
```
***
**Parent Module:** [`memory`](marvin-memory)
# providers
Source: https://askmarvin.ai/api-reference/marvin-memory-providers
# `marvin.memory.providers`
## Submodules
* [`chroma`](marvin-memory-providers-chroma)
* [`lance`](marvin-memory-providers-lance)
* [`postgres`](marvin-memory-providers-postgres)
* [`qdrant`](marvin-memory-providers-qdrant)
***
**Parent Module:** [`memory`](marvin-memory)
# chroma
Source: https://askmarvin.ai/api-reference/marvin-memory-providers-chroma
# `marvin.memory.providers.chroma`
## Classes
### `ChromaMemory`
```python theme={null}
class ChromaMemory(client: chromadb.ClientAPI = (lambda: chromadb.PersistentClient(path=(str(marvin.settings.home_path / 'memory/chroma'))))(), collection_name: str = 'memory-{key}')
```
**Methods:**
* **`add`**
```python theme={null}
def add(self, memory_key: str, content: str) -> str
```
* **`delete`**
```python theme={null}
def delete(self, memory_key: str, memory_id: str)
```
* **`get_collection`**
```python theme={null}
def get_collection(self, memory_key: str) -> chromadb.Collection
```
* **`search`**
```python theme={null}
def search(self, memory_key: str, query: str, n: int = 20) -> dict[str, str]
```
## Functions
### `ChromaCloudMemory`
```python theme={null}
def ChromaCloudMemory(tenant: str | None = None, database: str | None = None, api_key: str | None = None, **kwargs: Any = {}) -> ChromaMemory
```
### `ChromaEphemeralMemory`
```python theme={null}
def ChromaEphemeralMemory(**kwargs: Any = {}) -> ChromaMemory
```
### `ChromaPersistentMemory`
```python theme={null}
def ChromaPersistentMemory(path: str | None = None, **kwargs: Any = {}) -> ChromaMemory
```
***
**Parent Module:** [`providers`](marvin-memory-providers)
# lance
Source: https://askmarvin.ai/api-reference/marvin-memory-providers-lance
# `marvin.memory.providers.lance`
## Classes
### `LanceMemory`
```python theme={null}
class LanceMemory(uri: Path = marvin.settings.home_path / 'memory' / 'lancedb', table_name: str = 'memory-{key}', embedding_fn: Callable[..., Any] = (lambda: get_registry().get('openai').create(name='text-embedding-ada-002'))(), _cached_model: LanceModel | None = None)
```
**Methods:**
* **`add`**
```python theme={null}
def add(self, memory_key: str, content: str) -> str
```
* **`delete`**
```python theme={null}
def delete(self, memory_key: str, memory_id: str)
```
* **`get_db`**
```python theme={null}
def get_db(self) -> lancedb.DBConnection
```
* **`get_model`**
```python theme={null}
def get_model(self) -> LanceModel
```
* **`get_table`**
```python theme={null}
def get_table(self, memory_key: str) -> lancedb.table.Table
```
* **`search`**
```python theme={null}
def search(self, memory_key: str, query: str, n: int = 20) -> dict[str, str]
```
***
**Parent Module:** [`providers`](marvin-memory-providers)
# qdrant
Source: https://askmarvin.ai/api-reference/marvin-memory-providers-qdrant
# `marvin.memory.providers.qdrant`
## Classes
### `QdrantMemory`
```python theme={null}
class QdrantMemory(client: AsyncQdrantClient = (lambda: AsyncQdrantClient(path=(str(marvin.settings.home_path / 'memory/qdrant'))))(), collection_name: str = 'memory-{key}')
```
**Methods:**
* **`add`**
```python theme={null}
def add(self, memory_key: str, content: str) -> str
```
* **`delete`**
```python theme={null}
def delete(self, memory_key: str, memory_id: str)
```
* **`get_collection`**
```python theme={null}
def get_collection(self, memory_key: str) -> models.CollectionInfo
```
* **`search`**
```python theme={null}
def search(self, memory_key: str, query: str, n: int = 20) -> dict[str, str]
```
***
**Parent Module:** [`providers`](marvin-memory-providers)
# task
Source: https://askmarvin.ai/api-reference/marvin-tasks-task
Tasks for Marvin.
# `marvin.tasks.task`
Tasks for Marvin.
A Task is a container for a prompt and its associated state.
## Constants
### `NOTSET`
```python theme={null}
NOTSET = "'__NOTSET__'"
```
### `T`
```python theme={null}
T = TypeVar('T')
```
## Classes
### `Task`
```python theme={null}
class Task(instructions: str | Sequence[UserContent], attachments: Sequence[UserContent] | None = None, result_type: ResultType[T] = NOTSET, name: str | None = None, prompt_template: str | Path = Path('task.jinja'), agents: Actor | Sequence[Actor] | None = None, context: dict[str, Any] | None = None, tools: list[Callable[..., Any]] | None = None, memories: list[Memory] | None = None, result_validator: Callable[..., Any] | None = None, parent: Task[Any] | None | Literal['__NOTSET__'] = NOTSET, depends_on: Sequence[Task[Any]] | None = None, allow_fail: bool = False, allow_skip: bool = False, verbose: bool = False, cli: bool = False, plan: bool = False)
```
A task is a container for a prompt and its associated state.
**Methods:**
* **`friendly_name`**
```python theme={null}
def friendly_name(self, verbose: bool = True) -> str
```
Get a friendly name for this task.
* **`get_actor`**
```python theme={null}
def get_actor(self) -> Actor
```
Retrieve the actor assigned to this task.
* **`get_end_turn_tools`**
```python theme={null}
def get_end_turn_tools(self) -> list[type[EndTurn]]
```
Get the result tool for this task.
* **`get_prompt`**
```python theme={null}
def get_prompt(self) -> str | Sequence[UserContent]
```
Get the rendered prompt for this task.
Uses the task's prompt\_template (or default if None) and renders it with
this task instance as the `task` variable.
* **`get_result_type`**
```python theme={null}
def get_result_type(self) -> type[T]
```
Get the effective result type for this task.
For classification tasks, returns the type that should be used
for validation (e.g., int or list\[int]).
* **`get_result_type_str`**
```python theme={null}
def get_result_type_str(self) -> str
```
Get a string representation of the result type.
* **`get_tools`**
```python theme={null}
def get_tools(self) -> list[Callable[..., Any]]
```
Get the tools assigned to this task.
* **`is_classifier`**
```python theme={null}
def is_classifier(self) -> bool
```
Return True if this task is a classification task.
* **`is_complete`**
```python theme={null}
def is_complete(self) -> bool
```
Check if the task is complete.
* **`is_failed`**
```python theme={null}
def is_failed(self) -> bool
```
Check if the task is failed.
* **`is_incomplete`**
```python theme={null}
def is_incomplete(self) -> bool
```
Check if the task is incomplete.
* **`is_pending`**
```python theme={null}
def is_pending(self) -> bool
```
Check if the task is pending.
* **`is_ready`**
```python theme={null}
def is_ready(self) -> bool
```
Check if the task is ready to run.
A task is ready if it is incomplete and all of its dependencies (including subtasks) are complete.
* **`is_running`**
```python theme={null}
def is_running(self) -> bool
```
Check if the task is running.
* **`is_skipped`**
```python theme={null}
def is_skipped(self) -> bool
```
Check if the task is skipped.
* **`is_successful`**
```python theme={null}
def is_successful(self) -> bool
```
Check if the task is successful.
* **`mark_failed`**
```python theme={null}
def mark_failed(self, error: str, thread: Thread | None = None)
```
Mark the task as failed with an error message.
* **`mark_failed_tool`**
```python theme={null}
def mark_failed_tool(self) -> type[marvin.engine.end_turn.MarkTaskFailed]
```
* **`mark_running`**
```python theme={null}
def mark_running(self, thread: Thread | None = None)
```
Mark the task as running.
* **`mark_skipped`**
```python theme={null}
def mark_skipped(self, thread: Thread | None = None)
```
Mark the task as skipped.
* **`mark_skipped_tool`**
```python theme={null}
def mark_skipped_tool(self) -> type[marvin.engine.end_turn.MarkTaskSkipped]
```
* **`mark_successful`**
```python theme={null}
def mark_successful(self, result: T = None, validate_result: bool = True, thread: Thread | None = None)
```
Mark the task as successful with an optional result.
* **`mark_successful_tool`**
```python theme={null}
def mark_successful_tool(self) -> type[marvin.engine.end_turn.MarkTaskSuccessful]
```
* **`run`**
```python theme={null}
def run(self, thread: Thread | str | None = None, raise_on_failure: bool = True, handlers: list[Handler | AsyncHandler] | None = None) -> T
```
* **`run_async`**
```python theme={null}
def run_async(self, thread: Thread | str | None = None, raise_on_failure: bool = True, handlers: list[Handler | AsyncHandler] | None = None) -> T
```
* **`run_stream`**
```python theme={null}
def run_stream(self, thread: Thread | str | None = None, raise_on_failure: bool = True, handlers: list[Handler | AsyncHandler] | None = None) -> AsyncGenerator[Event, None]
```
* **`validate_result`**
```python theme={null}
def validate_result(self, raw_result: Any) -> T
```
Validate a result against the expected type and custom validator.
### `TaskState`
```python theme={null}
class TaskState()
```
State of a task.
## Functions
### `get_type_adapter`
```python theme={null}
def get_type_adapter(result_type: type[T]) -> TypeAdapter[T]
```
***
**Parent Module:** [`tasks`](marvin-tasks)
# thread
Source: https://askmarvin.ai/api-reference/marvin-tools-thread
# `marvin.tools.thread`
## Functions
### `post_message_to_agents`
```python theme={null}
def post_message_to_agents(message: str) -> str
```
Post a message to the thread for other agents to see. Do not use this to respond to the user.
***
**Parent Module:** [`tools`](marvin-tools)
# asyncio
Source: https://askmarvin.ai/api-reference/marvin-utilities-asyncio
# `marvin.utilities.asyncio`
## Constants
### `T`
```python theme={null}
T = TypeVar('T')
```
## Functions
### `run_sync`
```python theme={null}
def run_sync(coro: Coroutine[Any, Any, T]) -> T
```
Run a coroutine synchronously.
This function uses asyncio to run a coroutine in a synchronous context.
It attempts the following strategies in order:
1. If no event loop is running, creates a new one and runs the coroutine
2. If a loop is running, attempts to run the coroutine on that loop
3. As a last resort, creates a new thread with its own event loop to run the coroutine
Context variables are properly propagated between threads in all cases.
Example:
```python theme={null}
async def f(x: int) -> int:
return x + 1
result = run_sync(f(1))
```
Args:
coro: The coroutine to run synchronously
Returns:
The result of the coroutine
### `run_sync_in_thread`
```python theme={null}
def run_sync_in_thread(coro: Coroutine[Any, Any, T]) -> T
```
Run a coroutine synchronously in a new thread.
This function creates a new thread with its own event loop to run the coroutine.
Context variables are properly propagated between threads.
This is useful when you need to run async code in a context where you can't use
the current event loop (e.g., inside an async frame).
Example:
```python theme={null}
async def f(x: int) -> int:
return x + 1
result = run_sync_in_thread(f(1))
```
Args:
coro: The coroutine to run synchronously
Returns:
The result of the coroutine
***
**Parent Module:** [`utilities`](marvin-utilities)
# jinja
Source: https://askmarvin.ai/api-reference/marvin-utilities-jinja
# `marvin.utilities.jinja`
*No public API documentation found for this module.*
***
**Parent Module:** [`utilities`](marvin-utilities)
# jsonschema
Source: https://askmarvin.ai/api-reference/marvin-utilities-jsonschema
Convert JSON Schema to Python types with validation.
# `marvin.utilities.jsonschema`
Convert JSON Schema to Python types with validation.
The jsonschema\_to\_type function converts a JSON Schema into a Python type that can be used
for validation with Pydantic. It supports:
* Basic types (string, number, integer, boolean, null)
* Complex types (arrays, objects)
* Format constraints (date-time, email, uri)
* Numeric constraints (minimum, maximum, multipleOf)
* String constraints (minLength, maxLength, pattern)
* Array constraints (minItems, maxItems, uniqueItems)
* Object properties with defaults
* References and recursive schemas
* Enums and constants
* Union types
Example:
```python theme={null}
schema = {
"type": "object",
"properties": {
"name": {"type": "string", "minLength": 1},
"age": {"type": "integer", "minimum": 0},
"email": {"type": "string", "format": "email"}
},
"required": ["name", "age"]
}
# Name is optional and will be inferred from schema's "title" property if not provided
Person = jsonschema_to_type(schema)
# Creates a validated dataclass with name, age, and optional email fields
```
## Constants
### `FORMAT_TYPES`
```python theme={null}
FORMAT_TYPES = {'date-time': datetime, 'email': EmailStr, 'uri': AnyUrl, 'json': Json}
```
## Classes
### `JSONSchema`
```python theme={null}
class JSONSchema()
```
## Functions
### `create_array_type`
```python theme={null}
def create_array_type(schema: Mapping[str, Any], schemas: Mapping[str, Any]) -> type | Annotated[Any, ...]
```
Create list/set type with optional constraints.
### `create_dataclass`
```python theme={null}
def create_dataclass(schema: Mapping[str, Any], name: str | None = None, schemas: Mapping[str, Any] | None = None) -> type
```
Create dataclass from object schema.
### `create_enum`
```python theme={null}
def create_enum(name: str, values: list[Any]) -> type | Enum
```
Create enum type from list of values.
### `create_field_with_default`
```python theme={null}
def create_field_with_default(field_type: type, default_value: Any, schema: dict[str, Any]) -> Any
```
Create a field with simplified default handling.
### `create_numeric_type`
```python theme={null}
def create_numeric_type(base: Type[int | float], schema: Mapping[str, Any]) -> type | Annotated[Any, ...]
```
Create numeric type with optional constraints.
### `create_string_type`
```python theme={null}
def create_string_type(schema: Mapping[str, Any]) -> type | Annotated[Any, ...]
```
Create string type with optional constraints.
### `get_default_value`
```python theme={null}
def get_default_value(schema: dict[str, Any], prop_name: str, parent_default: dict[str, Any] | None = None) -> Any
```
Get default value with proper priority ordering.
1. Value from parent's default if it exists
2. Property's own default if it exists
3. None
### `hash_schema`
```python theme={null}
def hash_schema(schema: Mapping[str, Any]) -> str
```
Generate a deterministic hash for schema caching.
### `jsonschema_to_type`
```python theme={null}
def jsonschema_to_type(schema: Mapping[str, Any], name: str | None = None) -> type
```
Convert JSON schema to appropriate Python type with validation.
Args:
schema: A JSON Schema dictionary defining the type structure and validation rules
name: Optional name for object schemas. Only allowed when schema type is "object".
If not provided for objects, name will be inferred from schema's "title"
property or default to "Root".
Returns:
A Python type (typically a dataclass for objects) with Pydantic validation
Raises:
ValueError: If a name is provided for a non-object schema
Examples:
Create a dataclass from an object schema:
```python theme={null}
schema = {
"type": "object",
"title": "Person",
"properties": {
"name": {"type": "string", "minLength": 1},
"age": {"type": "integer", "minimum": 0},
"email": {"type": "string", "format": "email"}
},
"required": ["name", "age"]
}
Person = jsonschema_to_type(schema)
# Creates a dataclass with name, age, and optional email fields:
# @dataclass
# class Person:
# name: str
# age: int
# email: str | None = None
```
Person(name="John", age=30)
Create a scalar type with constraints:
```python theme={null}
schema = {
"type": "string",
"minLength": 3,
"pattern": "^[A-Z][a-z]+$"
}
NameType = jsonschema_to_type(schema)
# Creates Annotated[str, StringConstraints(min_length=3, pattern="^[A-Z][a-z]+$")]
@dataclass
class Name:
name: NameType
```
### `merge_defaults`
```python theme={null}
def merge_defaults(data: Mapping[str, Any], schema: Mapping[str, Any], parent_default: Mapping[str, Any] | None = None) -> dict[str, Any]
```
Merge defaults with provided data at all levels.
### `resolve_ref`
```python theme={null}
def resolve_ref(ref: str, schemas: Mapping[str, Any]) -> Mapping[str, Any]
```
Resolve JSON Schema reference to target schema.
### `sanitize_name`
```python theme={null}
def sanitize_name(name: str) -> str
```
Convert string to valid Python identifier.
### `schema_to_type`
```python theme={null}
def schema_to_type(schema: Mapping[str, Any], schemas: Mapping[str, Any]) -> type
```
Convert schema to appropriate Python type.
***
**Parent Module:** [`utilities`](marvin-utilities)
# logging
Source: https://askmarvin.ai/api-reference/marvin-utilities-logging
# `marvin.utilities.logging`
## Functions
### `get_logger`
```python theme={null}
def get_logger(name: str | None = None) -> logging.Logger
```
Retrieves a logger with the given name, or the root logger if no name is given.
Args:
name: The name of the logger to retrieve.
Returns:
The logger with the given name, or the root logger if no name is given.
Example:
Basic Usage of `get_logger`
```python theme={null}
from marvin.utilities.logging import get_logger
logger = get_logger("marvin.test")
logger.info("This is a test") # Output: marvin.test: This is a test
debug_logger = get_logger("marvin.debug")
debug_logger.debug_kv("TITLE", "log message", "green")
```
### `maybe_quote`
```python theme={null}
def maybe_quote(value: Any) -> str
```
### `setup_logging`
```python theme={null}
def setup_logging(settings: marvin.settings.Settings)
```
***
**Parent Module:** [`utilities`](marvin-utilities)
# models
Source: https://askmarvin.ai/api-reference/marvin-utilities-models
Base models for Marvin.
# `marvin.utilities.models`
Base models for Marvin.
## Classes
### `MarvinModel`
```python theme={null}
class MarvinModel()
```
Base model for all Marvin models.
This model:
* Is strict (no extra fields allowed)
* Uses modern Pydantic v2 features
***
**Parent Module:** [`utilities`](marvin-utilities)
# testing
Source: https://askmarvin.ai/api-reference/marvin-utilities-testing
# `marvin.utilities.testing`
## Functions
### `assert_llm_equal`
```python theme={null}
def assert_llm_equal(result: Any, expected: Any) -> bool
```
***
**Parent Module:** [`utilities`](marvin-utilities)
# tools
Source: https://askmarvin.ai/api-reference/marvin-utilities-tools
# `marvin.utilities.tools`
## Constants
### `T`
```python theme={null}
T = TypeVar('T')
```
## Classes
### `ResultTool`
```python theme={null}
class ResultTool(type: Literal['result-tool'] = 'result-tool')
```
**Methods:**
* **`run`**
```python theme={null}
def run(self, ctx: RunContext)
```
## Functions
### `update_fn`
```python theme={null}
def update_fn(func: Callable[..., T] | None = None, name: str | None = None, description: str | None = None) -> Callable[[Callable[..., T]], Callable[..., T]] | Callable[..., T]
```
Update a function's name and optionally set its docstring.
Can be used either as a decorator with keyword arguments or as a direct function.
Args:
func: The function to update (optional). If provided, updates are applied directly.
If not provided, returns a decorator.
name: The new name for the function (optional). If provided, must not be empty.
description: Optional docstring for the function
Example:
# As a function:
def my\_fn(x):
return x
updated\_fn = update\_fn(my\_fn, name='hello\_there')
# As a decorator with name:
@update\_fn(name='hello\_there')
def my\_fn(x):
return x
# As a decorator with name and description:
@update\_fn(name='hello\_there', description='Says hello')
def my\_fn(x):
return x
# As a decorator with no arguments:
@update\_fn()
def my\_fn(x):
return x
# Works with async functions too:
@update\_fn(name='async\_hello')
async def my\_async\_fn(x):
return x
### `wrap_tool_errors`
```python theme={null}
def wrap_tool_errors(tool_fn: Callable[..., Any])
```
Pydantic AI doesn't catch errors except for ModelRetry, so we need to make
sure we catch them ourselves and raise a ModelRetry instead.
***
**Parent Module:** [`utilities`](marvin-utilities)
# The Marvin Community
Source: https://askmarvin.ai/community
Join us!
![]()
We're thrilled you're interested in Marvin! Here, we're all about community. We're driven by a shared passion for making LLMs and agents more accessible and easier to use.
### Connect on Discord
Join us on [Discord](https://discord.gg/Kgw4HpcuYG) or [Slack](https://communityinviter.com/apps/prefect-community/prefect-community) to ask questions, share ideas, or just chat with like-minded developers. Don't be shy.
### Contributing to Marvin
We encourage all types of contributions, whether it be reporting bugs, suggesting features, or contributing code. Here's how you can contribute:
* **Issues:** Encountered a bug? Have a suggestion? Open an issue in our [GitHub repository](https://github.com/PrefectHQ/Marvin). We appreciate your input!
* **Pull Requests (PRs):** Ready to contribute code? We welcome your pull requests! Not sure how to make a PR? Check out the [GitHub guide](https://docs.github.com/en/github/collaborating-with-issues-and-pull-requests/about-pull-requests).
* **Discussions:** Have an idea but not quite ready to open an issue or PR? Create a GitHub Discussion.
* **Chat:** Join us in one of the above places to chat about ideas or behavior.
# Agents
Source: https://askmarvin.ai/concepts/agents
The intelligent workers in your AI workflows.
Agents are the intelligent, autonomous entities that power your AI workflows in Marvin. They represent AI models capable of understanding instructions, making decisions, and completing tasks.
```python theme={null}
import marvin
agent = marvin.Agent(name="Marvin")
```
## What are agents?
Agents in Marvin are configurable AI entities, each with its own identity, capabilities, and even personality. They act as the "workers" in your AI workflows, responsible for executing tasks and making decisions based on their assigned instructions and available tools.
You can think of each agent as a portable LLM configuration. When you assign an agent to a task, they will work to complete it according to the instructions and tools you provide.
## Creating agents
To create an agent, use the `Agent` class:
```python theme={null}
import marvin
agent = marvin.Agent(name="Marvin")
```
A more complex agent can be created by providing additional configuration. This example shows the main configuration options:
```python theme={null}
import marvin
agent = marvin.Agent(
name="Data Analyst",
description="An AI agent specialized in data analysis",
instructions="Perform data analysis tasks efficiently and accurately.",
tools=[analyze_data, plot_results],
model="openai:gpt-4o",
memories=[knowledge_base],
)
```
## Agent properties
### Name
An agent's name is an identifier that is visible to other agents in the workflow. It is used to distinguish between agents and for logging and debugging purposes. If no name is provided, one will be randomly chosen from a list of famous AI characters.
### Description
A description is a brief summary of the agent's role or specialization. This information is visible to other agents, and helps them understand the agent's capabilities and expertise.
### Instructions
Instructions are specific instructions or guidelines for the agent to follow during task execution. These instructions are private and not shared with other agents.
### Tools
Tools are Python functions that the agent can call to perform specific actions or computations. They are defined as a list of functions when creating an agent, and can be used to enhance the agent's capabilities. The agent will have access to these tools in every task they are assigned to. If a task defines additional tools, the agent will have access to those as well.
### Model
Each agent has a model, which is the LLM that powers the agent responses. This allows you to choose the most suitable model for your needs, based on factors such as performance, latency, and cost.
Marvin supports any model that is compatible with Pydantic AI. You can specify models using a string format like "openai:gpt-4" or by providing a Model instance:
```python theme={null}
import marvin
from pydantic_ai.models import OpenAIModel
# Using a string identifier
agent1 = marvin.Agent(model="openai:gpt-4")
# Using a model instance
model = OpenAIModel(name="gpt-4")
agent2 = marvin.Agent(model=model)
```
### Model Settings
You can customize the behavior of an agent's model using the `model_settings` parameter. This allows you to configure settings like temperature, top\_p, etc:
```python theme={null}
import marvin
agent = marvin.Agent(
name="Creative Writer",
model="openai:gpt-4o",
model_settings={"temperature": 0.8}
)
```
### Memories
Agents can be configured with memories that provide additional context or knowledge. These memories persist across tasks and can be used to store and retrieve relevant information:
```python theme={null}
import marvin
from marvin.memory import Memory
knowledge_base = Memory(key="documentation")
agent = marvin.Agent(
name="Support Agent",
memories=[knowledge_base]
)
```
## Assigning agents to tasks
Agents must be assigned to tasks in order to work on them. You can assign agents by passing them to the `agents` parameter when creating a task:
```python theme={null}
import marvin
agent = marvin.Agent(name="Poet")
task = marvin.Task(
instructions="Write a poem about AI",
agents=[agent]
)
```
If no agent is specified, Marvin will use the default agent configured in your settings.
You can also use the agent's `say` method for simple conversational interactions:
```python theme={null}
import marvin
agent = marvin.Agent(name="Assistant")
response = agent.say("What is the meaning of life?")
print(response)
```
# Tasks
Source: https://askmarvin.ai/concepts/tasks
The building blocks of AI workflows.
Tasks are the fundamental building blocks of AI workflows in Marvin. They represent discrete, well-defined objectives that need to be accomplished by an AI agent.
```python theme={null}
import marvin
task = marvin.Task("Write the Marvin docs")
```
## What are tasks?
LLMs excel when given clear, specific instructions that allow them to focus their knowledge and capabilities on a well-defined goal. A `Task` in Marvin is a structured way to define these instructions and guide the AI's behavior. Each task represents a "checkpoint" that requires the AI to meet an observable, verifiable goal. In this way, tasks serve as a bridge between the structured world of traditional software and the more fluid, adaptive world of AI.
This task-centric approach allows you to leverage the full power of AI while maintaining precise oversight. Each task becomes a checkpoint where you can validate outputs, ensuring that the AI's work aligns with your application's requirements and constraints.
## Creating tasks
A task in Marvin typically consists of:
* Instructions: what needs to be accomplished
* Expected output: what form the result should have, including any constraints or validation
* Agent: the AI entity responsible for executing the task
* Tools: any additional capabilities needed to complete the task
* Memories: any relevant context or history that should be available to the agent
In practice, you will often use the `marvin.run` function to create and run a task in a single step. This is a common operation and accepts all the same arguments as creating a `Task` directly.
### Using the `Task` class
The most straightforward way to create a task is by using the `Task` class:
```python Code theme={null}
import marvin
task = marvin.Task(
instructions="Write a poem about the provided topic",
result_type=str,
context={"topic": "AI"}
)
result = task.run()
print(result)
```
```text Result theme={null}
In circuits deep and code profound,
An AI's mind begins to sound.
Electric thoughts and data streams,
Crafting worlds and shaping dreams.
```
### Agents
The `agents` parameter specifies which AI agents are assigned to complete the task. If not provided, Marvin will use the default agent. You can provide either a single agent or a team of agents that will work together as a unit.
```python theme={null}
import marvin
task = marvin.Task(
instructions="Write a poem about AI",
agents=[marvin.Agent(name="Marvin", model="openai:gpt-4o-mini")]
)
```
## Task properties
When creating a task, you can configure various properties to define its behavior and requirements. Here are the key configuration options:
### Instructions
The instructions of a task tell the agent what needs to be accomplished. This is the only required task configuration, as it indicates what the agent should do.
```python Code theme={null}
import marvin
poem = marvin.run(instructions="Write a poem about AI")
print(poem)
```
```text Result theme={null}
In circuits deep and code profound,
An AI's mind begins to sound.
Electric thoughts and data streams,
Crafting worlds and shaping dreams.
```
Instructions can be "meta", especially if you have an agent that is working on your workflow itself (e.g. monitoring progress, creating new tasks, etc.). Be creative!
```python Code theme={null}
import marvin
marvin.run("Write a poem... then fail this task.", allow_fail=True)
```
```text Result theme={null}
ValueError: Task 04561cda ("Write a poem... then fail this task.")
failed: Task instructed to be marked as failed despite
no technical error
```
### Result type
A task's result type indicates the type of value that the task will return. This is used to validate the task's result and to help agents understand the task's output.
A variety of different result types are supported, including:
* Builtin types: `str`, `int`, `bool`, `list`, `dict`, etc.
* `None`: sometimes a task requires agents to take actions but not return any specific value
* Builtin collections: `Tuple[int, str]`, `List[str]`, `Dict[str, int]`, etc.
* Annotated types: `Annotated[str, "a 5 digit zip code"]`
* Pydantic models
* Lists of literal values: provide a list of values to require the agent to choose one of them as its result. For example,`["book", "movie", "album"]` would require the agent to choose one of the three values.
The default result type is `str`.
Pydantic model example:
```python Code theme={null}
import marvin
from pydantic import BaseModel
class Name(BaseModel):
first: str
last: str
name = marvin.run("The input is 'John Doe'", result_type=Name)
print(repr(name))
```
```text Result theme={null}
Name(first='John', last='Doe')
```
Classification example:
```python Code theme={null}
import marvin
media = marvin.run(
"Star Wars: Return of the Jedi",
result_type=["book", "movie", "album"]
)
print(media)
```
```text Result theme={null}
movie
```
### Result validator
You can specify a custom validation function for the task's result using the task's `result_validator` parameter. This function will be called with the raw result and should return the validated result or raise an exception if the result is not valid.
```python theme={null}
import marvin
def validate_even(value: int) -> int:
if value % 2 != 0:
raise ValueError("Value must be even")
return value
number = marvin.run("Choose a number", result_validator=validate_even)
print(number)
```
### Context
The context of a task is a dictionary that provides additional information about the task. This information is visible to agents during execution, helping them understand the task they are working on.
```python Code theme={null}
import marvin
is_spam = marvin.run(
"Is this email spam?",
result_type=bool,
context=dict(email='You just won a million dollars!'),
)
print(is_spam)
```
```text Result theme={null}
True
```
### Tools
The tools of a task are a list of tools that the task requires. These tools will be made available to the agent during execution.
```python Code theme={null}
import marvin
import random
def roll_dice(n_dice: int):
return [random.randint(1, 6) for _ in range(n_dice)]
rolls = marvin.run(
"Roll 3 dice",
result_type=list[int],
tools=[roll_dice],
)
print(rolls)
```
```text Result theme={null}
[3, 1, 5]
```
### Memories
Tasks can be configured with a list of memories that provide additional context to the agent. These memories can include relevant history, knowledge bases, or other information that helps the agent complete the task.
```python theme={null}
import marvin
from marvin.memory import Memory
memory = Memory(key="documentation")
memory.add("Marvin is a powerful AI framework...")
task = marvin.Task(
instructions="Explain what Marvin is",
memories=[memory]
)
```
### Parent
Tasks can be configured with a parent task. Creating hierarchies of tasks can help agents understand the relationships between different tasks and to better understand the task they are working on.
### Task Planning
When a task is created with `plan=True`, the agent is given the ability to break down complex tasks into smaller, more manageable subtasks. This is particularly useful for handling large or complex objectives that benefit from being tackled incrementally.
The agent can create subtasks at any point during task execution, and these subtasks become part of the task hierarchy. Each subtask is treated as a dependency of the parent task, ensuring that all subtasks are completed before the parent task is considered finished.
```python theme={null}
import marvin
task = marvin.Task(
instructions="Write a comprehensive report about AI",
plan=True # Enable task planning
)
```
When planning is enabled, the agent can:
* Create one or more subtasks at any time
* Define clear, focused objectives for each subtask
* Establish dependencies between subtasks
* Monitor progress through the task hierarchy
This feature is especially valuable for:
* Breaking down complex tasks into manageable pieces
* Creating checkpoints in long-running processes
* Enabling parallel work on independent subtasks
* Maintaining clear progress tracking
### Depends on
Tasks can be configured with a list of tasks that they depend on. This information helps Marvin prioritize work and avoid running a task before its upstream dependencies are complete.
### Additional Options
Marvin tasks support several additional options to control their behavior:
* `allow_fail`: Whether the task is allowed to be marked as failed (default: False)
* `allow_skip`: Whether the task is allowed to be skipped (default: False)
* `cli`: Whether to enable CLI interaction tools for the agent (default: False)
* `plan`: Whether to enable task planning capabilities, allowing the agent to create subtasks to help complete the parent task (default: False)
## Runtime properties
The following properties of a `Task` are set during task execution, and can be examined as part of your workflow's logic.
### Status
The status of a task reflects whether an agent has started working on it, and what the ultimate outcome of that work was. Tasks are always created with a `PENDING` status, progress to a `RUNNING` status when the agent begins to work on it, and finally moves to one of a few completed statuses when the task is finished.
Agents use tools to mark tasks as `SUCCESSFUL` or `FAILED`. Successful tasks will also have a `result` property, which contains the task's final output. This is a value that satisfies the task's instructions and result type configuration. Failed tasks will have an error message as their result.
```python Code theme={null}
import marvin
task = marvin.Task("Write a poem about AI")
task.run()
print(task.status)
```
```text Result theme={null}
TaskStatus.SUCCESSFUL
```
In addition to checking the status explicitly, you can call a number of helper methods on the task:
| Method | Description |
| ----------------- | ------------------------------------------------------------------------------------------------------------------- |
| `is_pending()` | Returns `True` if the task is pending. |
| `is_running()` | Returns `True` if the task is running. |
| `is_successful()` | Returns `True` if the task is successful. |
| `is_failed()` | Returns `True` if the task is failed. |
| `is_skipped()` | Returns `True` if the task is skipped. |
| `is_complete()` | Returns `True` if the task is complete (either successful, failed, or skipped) |
| `is_incomplete()` | Returns `True` if the task is incomplete (either pending or running) |
| `is_ready()` | Returns `True` if the task is ready to be worked on (i.e. all dependencies are complete but the task is incomplete) |
### Result
When a task is completed successfully, its `result` property will contain the task's final output. This is a value that satisfies the task's instructions and result type configuration.
If a task fails, its `result` property will contain an error message describing the failure.
```python Code theme={null}
import marvin
task = marvin.Task("Write a poem about AI")
task.run()
print(task.result)
```
```text Result theme={null}
In circuits deep and code profound,
An AI's mind begins to sound.
Electric thoughts and data streams,
Crafting worlds and shaping dreams.
```
### Tools and Context
Tasks in Marvin can be enhanced with tools and context:
* **Tools** are Python functions that the task can use to perform actions (like API calls or data retrieval)
* **Context** is additional information provided to help the task succeed
For example:
```python theme={null}
import marvin
import datetime
def get_current_time() -> str:
"""Get the current date and time."""
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
task = marvin.Task(
instructions="Tell me the current time and create a greeting appropriate for that time of day",
tools=[get_current_time],
context={"user_name": "Alex"}
)
result = task.run()
print(result)
```
For a comprehensive guide on working with tools and context, see the [Tools and Context](/concepts/tools-and-context) documentation.
# Threads
Source: https://askmarvin.ai/concepts/threads
Manage conversation history and context
Threads in Marvin provide a powerful mechanism for managing conversation history and context across your AI workflows. Unlike ControlFlow's flows, threads in Marvin are focused on conversation management and persistence rather than workflow orchestration.
## Understanding Threads
A thread in Marvin represents a conversation context that can span multiple interactions. Each thread:
* Maintains its own conversation history
* Can be persisted to a database
* Can be referenced across different parts of your application
* Supports attaching memories for long-term information storage
## Creating and Using Threads
### Basic Thread Creation
The simplest way to create a thread is to instantiate the `Thread` class:
```python theme={null}
import marvin
thread = marvin.Thread()
# A new thread is created with a unique ID
```
### Specifying a Thread ID
You can create a thread with a specific ID, which is useful for resuming conversations:
```python theme={null}
thread = marvin.Thread(id="user_123_lesson")
# This thread will use the specified ID
```
### Using Threads with Tasks
Threads can be passed to tasks to maintain conversation context:
```python theme={null}
import marvin
thread = marvin.Thread()
task = marvin.Task(
instructions="Write a poem about AI",
result_type=str
)
# Run the task with the thread
result = task.run(thread=thread)
```
### Adding Messages
You can manually add messages to a thread:
```python theme={null}
# Add a user message
thread.add_user_message("Hello!")
# Add multiple messages
from marvin.engine.llm import UserMessage, AssistantMessage
thread.add_messages([
UserMessage(content="What's the weather?"),
AssistantMessage(content="It's sunny today!")
])
```
## Retrieving History
### Getting Messages
You can retrieve messages from a thread with various filters:
```python theme={null}
# Get all messages
messages = thread.get_messages()
# Get messages with filters
messages = thread.get_messages(
include_system_messages=False, # Exclude system messages
before=datetime.now(), # Messages before a timestamp
after=some_timestamp, # Messages after a timestamp
limit=10 # Limit number of messages
)
```
### Tracking LLM Usage
Threads also track LLM API usage:
```python theme={null}
# Get all LLM calls
llm_calls = thread.get_llm_calls()
# Get usage statistics
usage = thread.get_usage()
```
## Thread Context Management
Marvin provides context management for threads, making it easy to set the current thread:
```python theme={null}
with thread:
# This thread is now the current thread
task1.run() # Uses the thread automatically
task2.run() # Same thread is used
# Thread context is restored after the block
```
You can also get the current thread:
```python theme={null}
current_thread = marvin.get_current_thread()
```
## Parent-Child Relationships
Threads can have parent-child relationships:
```python theme={null}
parent_thread = marvin.Thread()
child_thread = marvin.Thread(parent_id=parent_thread.id)
```
## Database Integration
Threads are automatically persisted to a database when configured:
```python theme={null}
# Messages and LLM calls are automatically saved
thread.add_user_message("Save this message")
# Later, create a thread with the same ID to access history
same_thread = marvin.Thread(id=thread.id)
messages = same_thread.get_messages() # Includes saved messages
```
## Threads and Memory
While threads provide short-term conversation context within a session, [Memory](/concepts/memory) offers long-term persistence across multiple sessions. The two can work together:
```python theme={null}
import marvin
from marvin import Thread, Memory, Agent
# Create a memory for storing user preferences
user_preferences = Memory(key="user_preferences")
# Create an agent with this memory
assistant = Agent(
name="Personal Assistant",
instructions="You are a helpful assistant",
memories=[user_preferences]
)
# Use the agent in a thread
with Thread() as thread:
# The agent can access both thread context (recent conversation)
# and memory (long-term stored information)
assistant.run("Remember that I prefer dark mode")
assistant.run("What are my UI preferences?") # Will recall from memory
```
Threads maintain the flow of conversation, while Memory stores specific information that should persist across different threads and sessions. See the [Memory](/concepts/memory) documentation for more details on working with persistent memory.
# Tools and Context
Source: https://askmarvin.ai/concepts/tools-and-context
Extend AI capabilities with custom functions and additional information
Tools and context are two powerful ways to enhance Marvin's capabilities, allowing agents to perform actions beyond their built-in abilities and access additional information when completing tasks.
## Tools: Extending Agent Capabilities
Tools are Python functions that agents can call during task execution. They allow agents to:
* Access external systems and APIs
* Perform calculations or data processing
* Interact with files, databases, or other resources
* Execute custom business logic
```python theme={null}
import marvin
import requests
def search_web(query: str) -> str:
"""Search the web for information about the given query."""
response = requests.get(f"https://api.search.com?q={query}")
return response.json()["results"]
# Use the tool in a task
result = marvin.run(
"Research the latest developments in quantum computing",
tools=[search_web]
)
```
When an agent has access to a tool, it can decide when and how to use it based on the task requirements. The agent sees the tool's name, docstring, and signature, and can call it with appropriate arguments.
### Creating Effective Tools
The best tools are:
1. **Focused**: Each tool should do one thing well
2. **Well-documented**: Clear docstrings help the agent understand when and how to use the tool
3. **Type-hinted**: Type annotations help the agent understand what inputs and outputs to expect
```python theme={null}
from typing import Annotated
from pydantic import Field
# Define custom types with descriptions
ZipCode = Annotated[str, Field(description="A 5-digit US ZIP code")]
Temperature = Annotated[float, Field(description="Temperature in Fahrenheit")]
def get_weather(location: ZipCode) -> Temperature:
"""Get the current temperature for the specified ZIP code."""
# Implementation details...
return 72.5
```
### Providing Tools
Tools can be provided at different levels:
```python theme={null}
# To a specific task
task_result = marvin.run("Check the weather", tools=[get_weather])
# To an agent (available for all tasks the agent performs)
weather_agent = marvin.Agent(
name="WeatherBot",
instructions="You provide weather information",
tools=[get_weather, get_forecast]
)
# To a thread (available for all tasks in the thread)
with marvin.Thread() as thread:
thread.tools = [get_weather, get_forecast]
marvin.run("What's the weather like?")
marvin.run("Will it rain tomorrow?")
```
## Context: Providing Additional Information
Context is additional information you provide to a task or agent to help it generate better responses. Unlike tools (which are functions the agent can call), context is static data that informs the agent's understanding.
```python theme={null}
import marvin
# Provide context to a task
response = marvin.run(
"Summarize the patient's condition",
context={
"medical_records": "Patient presents with fever (101.2°F), cough, and fatigue...",
"patient_history": "42-year-old male with history of asthma...",
"lab_results": "White blood cell count elevated at 12,000..."
}
)
```
Context can include:
* Background information
* User preferences or history
* Data to analyze or reference
* Previous conversation history
* System state or configuration
### Using Context Effectively
For best results with context:
1. **Be selective**: Include only relevant information to avoid overwhelming the agent
2. **Structure appropriately**: Organize complex context into clear sections
3. **Use descriptive keys**: Name context variables clearly so the agent understands what each piece represents
```python theme={null}
# Example of well-structured context
context = {
"user_profile": {
"name": "Alex",
"preferences": ["science fiction", "technology", "hiking"],
"reading_level": "advanced"
},
"previous_recommendations": [
"The Three-Body Problem",
"Project Hail Mary"
]
}
recommendation = marvin.run(
"Recommend a book Alex might enjoy",
context=context
)
```
### Context in Prompts
You can reference context directly in your prompts using string formatting:
```python theme={null}
import marvin
user_data = {"name": "Jamie", "goal": "lose weight"}
plan = marvin.run(
"Create a fitness plan for {name} who wants to {goal}",
context=user_data
)
```
## Combining Tools and Context
Tools and context work together to create powerful AI workflows:
```python theme={null}
import marvin
import pandas as pd
def analyze_data(data: list[dict]) -> dict:
"""Analyze the provided data and return statistics."""
df = pd.DataFrame(data)
return {
"mean": df.mean().to_dict(),
"median": df.median().to_dict(),
"correlation": df.corr().to_dict()
}
# Combine tools and context
sales_data = [
{"product": "A", "price": 10, "units": 100},
{"product": "B", "price": 15, "units": 50},
# ...more data
]
analysis = marvin.run(
"Analyze our sales data and recommend pricing strategy",
tools=[analyze_data],
context={
"sales_data": sales_data,
"business_goal": "Increase revenue by 20% this quarter",
"competitor_prices": {"A": 12, "B": 14}
}
)
```
In this example:
* The context provides the data and business information
* The tool gives the agent the ability to perform statistical analysis
* Together, they enable the agent to make informed recommendations
## Best Practices
* **Security**: Only provide tools that you're comfortable with the agent using
* **Error handling**: Ensure tools handle errors gracefully and provide helpful error messages
* **Complexity**: Start with simple tools and context, then add complexity as needed
* **Testing**: Test your tools independently before providing them to agents
* **Observability**: Monitor tool usage to understand how agents are using them
By effectively combining tools and context, you can create AI workflows that leverage both the reasoning capabilities of large language models and the precision of traditional code.
# Cast
Source: https://askmarvin.ai/functions/cast
Convert data between types while preserving meaning
The `cast` function is your bridge between natural language and structured data types. It transforms `str → T` while preserving meaning, making it easy to convert:
* Natural numbers ("three point five" → 3.5)
* Casual responses ("sounds good" → True)
* Time expressions ("next Tuesday" → datetime)
* Rich descriptions ("John is 25" → Person(name="John", age=25))
For complex transformations, consider creating a custom task. The `cast` function is a convenient wrapper around Marvin's task system - see [Tasks](/concepts/tasks) for more details.
## Usage
Convert natural language descriptions of numbers into their numeric representation:
```python theme={null}
import marvin
# Convert text to a number
price = marvin.cast("three dollars and fifty cents", float)
print(price)
```
```python theme={null}
3.5
```
## Parameters
* `data`: The input data to convert (any type)
* `target`: The target type to convert to (defaults to `str`)
* `instructions`: Optional instructions to guide the conversion
* `agent`: Optional custom agent to use
* `thread`: Optional thread for conversation context
* `context`: Optional additional context
## Async Support
The function is also available in an async version:
```python theme={null}
import marvin
import asyncio
async def main():
result = await marvin.cast_async(
"three point five",
float
)
print(result) # 3.5
asyncio.run(main())
```
## Examples
### Converting Numbers
Convert natural language descriptions of quantities into numeric values:
```python theme={null}
import marvin
price = marvin.cast("three dollars and fifty cents", float)
print(price)
```
```python theme={null}
3.5
```
### Boolean Conversion
Interpret natural language expressions as boolean values:
```python theme={null}
import marvin
answer = marvin.cast("that sounds great!", bool)
print(answer)
```
```python theme={null}
True
```
### Structured Data
Extract structured information from natural language descriptions:
```python theme={null}
import marvin
from dataclasses import dataclass
@dataclass
class Address:
street: str
city: str
country: str
address = marvin.cast(
"123 Main St in New York, USA",
Address
)
print(f"{address.street}, {address.city}, {address.country}")
```
```python theme={null}
"123 Main St", "New York", "USA"
```
### Custom Instructions
Use instructions to control how text is interpreted and formatted:
```python theme={null}
from typing import Annotated
import marvin
from pydantic import Field
phone = marvin.cast(
"call me at 800-555-0123",
Annotated[str, Field(pattern=r"\d{1}-\d{3}-\d{3}-\d{4}")],
instructions="extract the phone number and assume the country code is 1"
)
print(phone)
```
```python theme={null}
"1-800-555-0123"
```
# Classify
Source: https://askmarvin.ai/functions/classify
Assign data to predefined categories
The `classify` function turns any input into clear categories. It transforms `str → L` where `L` is your set of labels, making it easy to identify:
* Sentiment ("Great product!" → Positive)
* Topics ("AI in healthcare" → \[Technology, Healthcare])
* Priority ("URGENT: System down" → Critical)
* Status ("All tests passing" → Success)
For complex classification needs, consider creating a custom task. The `classify` function is a convenient wrapper around Marvin's task system - see [Tasks](/concepts/tasks) for more details.
## Usage
Classify text into predefined categories:
```python theme={null}
import marvin
sentiment = marvin.classify(
"This product is amazing!",
["positive", "negative", "neutral"]
)
print(sentiment)
```
```python theme={null}
"positive"
```
## Parameters
* `data`: The input data to classify (any type)
* `labels`: Either a sequence of labels or an Enum class
* `multi_label`: Whether to return multiple labels (defaults to `False`)
* `instructions`: Optional instructions to guide classification
* `agent`: Optional custom agent to use
* `thread`: Optional thread for conversation context
* `context`: Optional additional context
## Async Support
The function is also available in an async version:
```python theme={null}
import marvin
import asyncio
async def main():
result = await marvin.classify_async(
"The weather is sunny and warm",
["sunny", "rainy", "cloudy"]
)
print(result) # "sunny"
asyncio.run(main())
```
## Examples
### Basic Classification
Use simple string labels to categorize content:
```python theme={null}
import marvin
priority = marvin.classify(
"Critical system failure",
["low", "medium", "high", "critical"]
)
print(priority)
```
```python theme={null}
"critical"
```
### Using Enums
Add type safety with Python enums:
```python theme={null}
import marvin
from enum import Enum
class Priority(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
priority = marvin.classify(
"Minor UI bug in non-critical component",
Priority
)
print(priority)
print(priority.value)
```
```python theme={null}
Priority.LOW
"low"
```
### Multi-label Classification
Identify multiple applicable categories:
```python theme={null}
import marvin
genres = marvin.classify(
"A romantic comedy about AI",
["comedy", "romance", "sci-fi", "drama", "action"],
multi_label=True
)
print(genres)
```
```python theme={null}
["comedy", "romance", "sci-fi"]
```
### Custom Instructions
Guide the classification with specific criteria:
```python theme={null}
import marvin
sentiment = marvin.classify(
"This product is okay",
["positive", "negative"],
instructions="Classify as positive only if explicitly enthusiastic"
)
print(sentiment)
```
```python theme={null}
"negative"
```
# Extract
Source: https://askmarvin.ai/functions/extract
Pull structured data from unstructured content
The `extract` function finds and pulls out specific pieces of information. It transforms `str → list[T]`, making it easy to collect:
* Contact details ("email: [alice@example.com](mailto:alice@example.com)" → \["[alice@example.com](mailto:alice@example.com)"])
* Measurements ("The room is 20.5 feet wide" → \[20.5])
* Dates ("Meeting on Jan 15th" → \[datetime(2024, 1, 15)])
* Structured data ("Alice (25) and Bob (30)" → \[Person(name="Alice", age=25), Person(name="Bob", age=30)])
For complex extraction patterns, consider creating a custom task. The `extract` function is a convenient wrapper around Marvin's task system - see [Tasks](/concepts/tasks) for more details.
## Usage
Pull email addresses from text:
```python theme={null}
import marvin
emails = marvin.extract(
"Contact us at support@example.com or sales@example.com",
str,
instructions="Find email addresses"
)
print(emails)
```
```python theme={null}
["support@example.com", "sales@example.com"]
```
## Parameters
* `data`: The input data to extract from (any type)
* `target`: The type of data to extract (defaults to `str`)
* `instructions`: Required when target is `str` to specify what to extract
* `agent`: Optional custom agent to use
* `thread`: Optional thread for conversation context
* `context`: Optional additional context
## Async Support
The function is also available in an async version:
```python theme={null}
import marvin
import asyncio
async def main():
result = await marvin.extract_async(
"The temperature is 72°F today",
float
)
print(result) # [72.0]
asyncio.run(main())
```
## Examples
### Numeric Values
Find all numbers in text:
```python theme={null}
import marvin
temperatures = marvin.extract(
"Today's high is 75°F with a low of 62°F",
float
)
print(temperatures)
```
```python theme={null}
[75.0, 62.0]
```
### Named Entities
Extract specific types of information:
```python theme={null}
import marvin
cities = marvin.extract(
"Flights from New York to London",
str,
instructions="Extract city names"
)
print(cities)
```
```python theme={null}
["New York", "London"]
```
### Dates and Times
Find temporal information in text:
```python theme={null}
import marvin
from datetime import datetime
dates = marvin.extract(
"The meeting is on March 15, 2024 at 2:30 PM",
datetime,
instructions="Extract date and time information"
)
print(dates)
```
```python theme={null}
[datetime(2024, 3, 15, 14, 30)]
```
### Structured Data
Pull complex information into structured types:
```python theme={null}
import marvin
from dataclasses import dataclass
@dataclass
class Contact:
name: str
email: str
phone: str | None
contacts = marvin.extract(
"""
John Smith (john@example.com, 555-0123)
Mary Johnson (mary@example.com)
""",
Contact
)
print(f"{contacts[0].name}: {contacts[0].phone}")
print(f"{contacts[1].name}: {contacts[1].phone}")
```
```python theme={null}
John Smith: 555-0123
Mary Johnson: None
```
# Generate
Source: https://askmarvin.ai/functions/generate
Create examples of any data type
The `generate` function creates realistic examples of any type. It transforms `() → T`, making it easy to produce:
* Test data ("Generate a plausible email address" → "[sarah.jones@company.com](mailto:sarah.jones@company.com)")
* Sample content ("Generate a product description" → "Sleek wireless earbuds with...")
* Structured examples ("Generate a recipe" → Recipe(title="Lemon Garlic Pasta", ingredients=\[...]))
For complex generation patterns, consider creating a custom task. The `generate` function is a convenient wrapper around Marvin's task system - see [Tasks](/concepts/tasks) for more details.
## Usage
Generate a simple string:
```python theme={null}
import marvin
name = marvin.generate(
str,
instructions="Generate a fantasy character name"
)
print(name)
```
```python theme={null}
"Thaelar Moonweaver"
```
## Parameters
* `target`: The type of data to generate
* `n`: Number of examples to generate (default: 1)
* `instructions`: Optional guidance for generation
* `agent`: Optional custom agent to use
* `thread`: Optional thread for conversation context
* `context`: Optional additional context
## Async Support
The function is also available in an async version:
```python theme={null}
import marvin
import asyncio
async def main():
result = await marvin.generate_async(
str,
instructions="Generate a company slogan"
)
print(result) # "Innovation meets simplicity"
asyncio.run(main())
```
## Examples
### Multiple Examples
Generate a list of items:
```python theme={null}
import marvin
colors = marvin.generate(
str,
n=3,
instructions="Generate color names"
)
print(colors)
```
```python theme={null}
["Cerulean Blue", "Crimson Red", "Forest Green"]
```
### Structured Data
Generate complex objects:
```python theme={null}
import marvin
from dataclasses import dataclass
from typing import List
@dataclass
class Recipe:
title: str
ingredients: List[str]
steps: List[str]
recipe = marvin.generate(Recipe)
print(f"Recipe: {recipe.title}")
print(f"First step: {recipe.steps[0]}")
```
```python theme={null}
"Recipe: Honey Glazed Salmon"
"First step: Preheat the oven to 400°F (200°C)"
```
### With Context
Generate content with specific requirements:
```python theme={null}
import marvin
description = marvin.generate(
str,
instructions="Generate a product description",
context="Target audience: tech-savvy professionals"
)
print(description)
```
```python theme={null}
"The XPS Pro 5000 ultrabook combines cutting-edge performance with enterprise-grade security features, perfect for remote work and intensive multitasking."
```
### Collections
Generate lists and dictionaries:
```python theme={null}
import marvin
# Generate a list of strings
ingredients = marvin.generate(list[str], instructions="Generate recipe ingredients")
print(ingredients)
# Generate a dictionary
menu = marvin.generate(dict[str, float], instructions="Generate menu items with prices")
print(menu)
```
```python theme={null}
["olive oil", "garlic", "pasta", "lemon", "parmesan cheese"]
{"Margherita Pizza": 12.99, "Caesar Salad": 8.99, "Tiramisu": 6.99}
```
# Run
Source: https://askmarvin.ai/functions/run
Execute AI tasks with a single line of code
The `run` function is the simplest way to execute AI tasks in Marvin. It provides a clean, one-line interface for running tasks while handling all the complexity of task creation and execution under the hood.
## Motivation
While Marvin provides powerful tools for complex AI workflows, many use cases are simple and don't need all that complexity. The `run` function provides:
* A simple, intuitive interface
* Sensible defaults
* Type safety through result types
* Easy access to common parameters
## Usage
```python theme={null}
import marvin
# Simple text generation
poem = marvin.run(
"Write a haiku about coding",
result_type=str
)
print(poem)
# Structured output
from pydantic import BaseModel
class Recipe(BaseModel):
name: str
ingredients: list[str]
steps: list[str]
recipe = marvin.run(
"Create a recipe for chocolate chip cookies",
result_type=Recipe
)
print(recipe.name)
print(recipe.ingredients)
```
```python theme={null}
Keys tap in silence,
Ideas flow through the night air,
Code becomes my art.
Chocolate Chip Cookies
['1 cup unsalted butter, softened', '1 cup white sugar', '1 cup packed brown sugar', '2 large eggs', '2 teaspoons vanilla extract', '3 cups all-purpose flour', '1 teaspoon baking soda', '1/2 teaspoon baking powder', '1/2 teaspoon salt', '2 cups semisweet chocolate chips']
```
## Parameters
* `instructions`: What you want the AI to do
* `result_type`: The expected type of the result (defaults to `str`)
* `tools`: Optional list of functions the AI can use
* `thread`: Optional thread for conversation context
* `agents`: Optional list of agents to use
* `raise_on_failure`: Whether to raise exceptions on failure (defaults to `True`)
* `handlers`: Optional list of handlers for events
* `**kwargs`: Additional context passed to the task
## Async Support
The function is also available in an async version:
```python theme={null}
import marvin
import asyncio
async def main():
# Run a task asynchronously
result = await marvin.run_async(
"Translate 'hello' to French",
result_type=str
)
print(result)
asyncio.run(main()) # bonjour
```
## Examples
### Simple Text Tasks
```python theme={null}
import marvin
# Generate text
story = marvin.run(
"Write a short story about a robot learning to paint"
)
print(story)
# Answer questions
answer = marvin.run(
"What is the capital of France?",
result_type=str
)
print(answer) # "Paris"
```
### Using Tools
```python theme={null}
import marvin
import random
def roll_die() -> int:
return random.randint(1, 6)
# Let the AI use tools
result = marvin.run(
"Roll a die and tell me if it's higher than 3",
result_type=bool,
tools=[roll_die]
)
print(result) # True or False depending on the roll
```
### Structured Output
```python theme={null}
import marvin
from pydantic import BaseModel, Field
class MovieReview(BaseModel):
title: str
rating: float = Field(gt=0, lt=5)
pros: list[str]
cons: list[str]
# Get structured review
review = marvin.run(
"Review the movie 'The Matrix'",
result_type=MovieReview
)
print(review.rating) # 4.8
print(review.pros) # ["Groundbreaking special effects", "Deep philosophical themes"]
```
### Using Context
```python theme={null}
import marvin
# Provide additional context
summary = marvin.run(
"Summarize this data in 2-3 sentences",
context={
"data": "Long text to summarize...",
"style": "technical"
}
)
print(summary)
```
### Multiple Tasks
The `run_tasks` function allows you to run multiple tasks at once:
```python theme={null}
import marvin
# Create tasks
task1 = marvin.Task(
instructions="Write a haiku",
result_type=str
)
task2 = marvin.Task(
instructions="Write a limerick",
result_type=str
)
# Run them together
tasks = marvin.run_tasks([task1, task2])
print(tasks[0].result) # Haiku
print(tasks[1].result) # Limerick
```
### Async Multiple Tasks
```python theme={null}
import marvin
import asyncio
async def main():
# Run multiple tasks asynchronously
tasks = await marvin.run_tasks_async(
[
marvin.Task("Write a haiku"),
marvin.Task("Write a limerick")
]
)
for task in tasks:
print(task.result)
asyncio.run(main())
```
# Say
Source: https://askmarvin.ai/functions/say
Add a message to a thread and get an agent's response
The `say` function adds a user message to a thread and gets an agent's response. It transforms `str → str`, making it easy to:
* Store messages in a thread ("My name is Alice" → thread.messages += \[user\_msg, agent\_msg])
* Get contextual responses ("What's my name?" → "Your name is Alice")
* Guide agent behavior ("You are a chef..." → "Let me suggest a recipe...")
For complex conversations, consider creating a custom task. The `say` function is a convenient wrapper around Marvin's task system - see [Tasks](/concepts/tasks) for more details.
## Usage
Add a message and get a response:
```python theme={null}
import marvin
# Create a thread to store the conversation
thread = marvin.Thread()
# Add a message and get a response
response = marvin.say(
"My name is Alice",
thread=thread # Messages are stored in the thread
)
print(response)
# The agent can now reference previous messages
response = marvin.say(
"What's my name?",
thread=thread
)
print(response)
```
```python theme={null}
"Hello Alice! Nice to meet you."
"Your name is Alice."
```
## Parameters
* `message`: The message to add to the thread
* `instructions`: Optional guidance for the agent's responses
* `agent`: Optional custom agent to use
* `thread`: Optional thread to store conversation history
* `context`: Optional additional context
## Async Support
The function is also available in an async version:
```python theme={null}
import marvin
import asyncio
async def main():
thread = marvin.Thread()
response = await marvin.say_async(
"What's the weather like?",
thread=thread
)
print(response)
asyncio.run(main())
```
## Examples
### With Instructions
Guide the agent's responses:
```python theme={null}
import marvin
thread = marvin.Thread()
response = marvin.say(
"What should I cook for dinner?",
instructions="Respond as a professional chef",
thread=thread
)
print(response)
```
```python theme={null}
"I'd recommend a pan-seared salmon with lemon butter sauce. Start by patting the salmon dry and seasoning with salt and pepper..."
```
### With Custom Agent
Use a specialized agent:
```python theme={null}
import marvin
thread = marvin.Thread()
tech_agent = marvin.Agent(
name="TechSupport",
instructions="You are a helpful tech support agent"
)
response = marvin.say(
"My computer won't start",
agent=tech_agent,
thread=thread
)
print(response)
```
```python theme={null}
"Let's troubleshoot this step by step. First, check if your computer is properly plugged in and the power button lights up..."
```
### With Memory
Remember previous interactions:
```python theme={null}
import marvin
thread = marvin.Thread()
agent = marvin.Agent(
memories=[
marvin.Memory(
key="user_preferences",
instructions="Remember user preferences"
)
]
)
# Store preference in memory
response = marvin.say(
"I prefer vegetarian food",
agent=agent,
thread=thread
)
print(response)
# Agent can access both memory and thread history
response = marvin.say(
"What should I cook?",
agent=agent,
thread=thread
)
print(response)
```
```python theme={null}
"I'll remember your preference for vegetarian food."
"Given your vegetarian preference, I'd suggest a delicious mushroom risotto..."
```
# Summarize
Source: https://askmarvin.ai/functions/summarize
Create concise summaries of any content
Summarizing text is such a common operation that we created a dedicated function for it. The
`summarize` function transforms any text into a concise summary, with optional instructions for
format and focus.
For complex summarization needs, consider creating a custom task. The `summarize` function is a
convenient wrapper around Marvin's task system - see [Tasks](/concepts/tasks) for more details.
## Usage
Create a simple summary:
```python theme={null}
import marvin
text = """
The annual baking competition featured thirty contestants this year, each bringing their
unique recipes and techniques. The highlight was a three-tiered chocolate cake with
raspberry filling, which won first place. Second place went to a creative take on
traditional apple pie, while third place was awarded to an innovative gluten-free
cheesecake.
"""
summary = marvin.summarize(text)
print(summary)
```
```python theme={null}
"A baking competition with 30 contestants saw a chocolate raspberry cake win first place, followed by an apple pie and a gluten-free cheesecake."
```
## Parameters
* `data`: The content to summarize
* `instructions`: Optional guidance for the summary
* `agent`: Optional custom agent to use
* `thread`: Optional thread for conversation context
* `context`: Optional dict of additional context
## Async Support
The function is also available in an async version:
```python theme={null}
import marvin
import asyncio
async def main():
result = await marvin.summarize_async(
"Long text here...",
instructions="Focus on key findings"
)
print(result)
asyncio.run(main())
```
## Examples
### Bullet Points
Get a structured summary:
```python theme={null}
import marvin
text = """
The new smartphone features a 6.7-inch OLED display, 5G connectivity,
and a triple-camera system. Battery life is rated at 24 hours of normal use.
The device runs on a custom chip that enables advanced AI features.
Pricing starts at $999 for the base model with 128GB storage.
"""
summary = marvin.summarize(
text,
instructions="Summarize very concisely in three bullet points (•)"
)
print(summary)
```
```python theme={null}
• The smartphone has a 6.7-inch OLED screen, 5G support, and a triple-camera system.
• It offers 24 hours of battery life and runs on a custom chip for advanced AI capabilities.
• Starting price is $999 for the 128GB model.
```
### With Context
Provide background information:
```python theme={null}
import marvin
text = """
Version 2.0 introduces several major changes. The API has been completely redesigned
for better usability. Several deprecated functions have been removed, and new features
have been added based on user feedback.
"""
summary = marvin.summarize(
text,
context={
"audience": "developers",
"document_type": "changelog"
},
instructions="Highlight breaking changes"
)
print(summary)
```
```python theme={null}
"Breaking changes include a complete API redesign and removal of deprecated functions, with new features added."
```
# Using MCP Servers
Source: https://askmarvin.ai/guides/mcp
Connect Marvin agents to Model Context Protocol (MCP) servers for extended capabilities.
MCP support is experimental and subject to change. [Open an issue on GitHub](https://github.com/PrefectHQ/marvin/issues) if you'd like to see a feature added or have any feedback.
Marvin agents can connect to [Model Context Protocol (MCP)](https://modelcontextprotocol.io/) servers, allowing them to access a wide array of external tools and data sources.
## What is MCP?
MCP provides a standardized way for AI agents to interface with tools, resources, prompts etc. If a thing in the world has an MCP server, you can trivially write an agent that interacts with it (on the terms of the MCP server implementation).
## Connecting an Agent to an MCP Server
To give your agent access to an MCP server, you simply pass `MCPServer` instances to the `mcp_servers` argument when creating your `Agent`.
Marvin uses `pydantic-ai` under the hood for MCP server interactions. The most common server type you'll likely use is `MCPServerStdio`, which communicates with an MCP server running as a subprocess via its standard input/output.
### Example: Using a Python Interpreter via MCP
Let's say you want your agent to be able to run Python code. There are MCP servers that provide a Python interpreter as a tool. The `jsr:@pydantic/mcp-run-python` package is one such server that can be run with Deno.
```python theme={null}
from marvin.agents import Agent
from pydantic_ai.mcp import MCPServerStdio
# Configure the MCP server for running Python code
# This requires Deno and the jsr:@pydantic/mcp-run-python package
# You can install the Deno package with:
# deno install -A -g jsr:@pydantic/mcp-run-python
run_python_mcp_server = MCPServerStdio(
command="deno", # Or the full path to your deno executable
args=["run", "-A", "jsr:@pydantic/mcp-run-python", "stdio"],
)
# Create an agent and provide the MCP server
coder_agent = Agent(
name="Coder",
instructions="Use the Python interpreter to solve tasks.",
mcp_servers=[run_python_mcp_server]
)
async def run_code():
result = await coder_agent.run_async(
"Use python to calculate the square root of 256 and tell me the result."
)
print(result)
if __name__ == "__main__":
asyncio.run(run_code())
```
Make sure the command specified in `MCPServerStdio` (e.g., `"deno"` or `"uvx"`) is available in your system's PATH or provide the full path to the executable.
### Example: Using a Git Tool Server via MCP
Another common use case is interacting with Git repositories. The `mcp-server-git` package (runnable with `uvx`, part of `uv`) provides Git tools over MCP.
```python theme={null}
from marvin.agents import Agent
from pydantic_ai.mcp import MCPServerStdio
# Configure the MCP server for Git tools
# This requires uv and mcp-server-git (installable via: uvx mcp-server-git, if needed)
git_mcp_server = MCPServerStdio(
command="uvx",
args=["mcp-server-git"]
)
version_control_agent = Agent(
name="VersionControlPro",
instructions="Use Git tools to inspect the repository.",
mcp_servers=[git_mcp_server]
)
async def get_repo_info():
result = await version_control_agent.run_async(
"Get the latest commit message from this repository."
)
print(result)
if __name__ == "__main__":
asyncio.run(get_repo_info())
```
## How it Works
tldr: It's basically just function calling (for now)
When an agent configured with `mcp_servers` runs:
1. Marvin discovers the tools provided by each active MCP server.
2. These discovered tools (e.g., `run_python_code` from the Python server, or `git_log` from the Git server) are made available to the LLM just like standard Python tools.
3. If the LLM decides to use an MCP tool, Marvin's engine handles the communication with the specified `MCPServer` instance.
4. The result from the MCP server is processed and returned to the LLM, which then formulates the final response.
## Exploring Further
For a more comprehensive example demonstrating multiple MCP servers and tools, check out the example script in the Marvin repository:
[`examples/agent_mcp.py`](https://github.com/PrefectHQ/marvin/blob/main/examples/agent_mcp.py)
This example showcases an agent that uses both a Python interpreter and Git tools via MCP to perform a multi-step task. You can run it locally to see MCP integration in action.
First, ensure you have the necessary MCP servers installed. For `examples/agent_mcp.py`, this typically means having Deno and `jsr:@pydantic/mcp-run-python` available, as well as `uv` and `mcp-server-git` (if you intend to use the git server part).
Then, from the root of the Marvin repository, you can run:
```bash theme={null}
uv run examples/agent_mcp.py
```
You should see the agent using tools like `git_log` and `run_python_code` to complete its task.
This setup allows Marvin agents to flexibly leverage a growing ecosystem of MCP-compatible tools and services.
```python theme={null}
```
# Installation
Source: https://askmarvin.ai/installation
Get started with Marvin in under a minute
## Requirements
* Python 3.10 or higher
* An API key from an LLM provider (OpenAI by default)
## Install `marvin`
The easiest way to install Marvin is a package manager like `pip` or [`uv`](https://docs.astral.sh/uv/):
```bash uv theme={null}
# add marvin to your project
uv add marvin
# add marvin to your script
uv add --script some_script.py marvin
# run a python process with marvin installed ephemerally
uv run --with marvin some_script.py
```
```bash pip theme={null}
pip install marvin
```
## Configure your LLM provider
By default, Marvin uses OpenAI's models. Set your API key as an environment variable:
```bash theme={null}
export OPENAI_API_KEY="your-api-key"
```
To use another provider, see the docs on [configuring LLMs](/guides/configure-llms).
## Development Installation
For development, install Marvin with development dependencies:
```bash uv theme={null}
git clone https://github.com/prefecthq/marvin.git
cd marvin
uv venv && source .venv/bin/activate
uv sync --dev
```
```bash pip theme={null}
git clone https://github.com/prefecthq/marvin.git
cd marvin
python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
```
## What's Next?
* Follow the [Quickstart](/quickstart) guide to build your first AI application
* Learn about [Tasks](/concepts/tasks), the building blocks of AI workflows
* Explore [Agents](/concepts/agents) to create specialized AI workers
* Read about [Threads](/concepts/threads) for managing conversation context
# Interactivity
Source: https://askmarvin.ai/patterns/interactivity
Enable agents to interact with users through the CLI.
Marvin allows agents to interact with users through the command line interface (CLI). This is useful for:
* Gathering user input during task execution
* Creating interactive chatbots
* Building conversational interfaces
* Getting user confirmation or choices
* Providing real-time feedback
Note that this is not a replacement for a proper user interface, and is intended primarily for local prototyping and development.
## Enabling CLI Interaction
To enable CLI interaction for a task, set the `cli` parameter to `True`:
```python theme={null}
import marvin
# Get user input
name = marvin.run(
"Ask for the user's name",
cli=True
)
print(f"The user's name is: {name}")
```
## Interactive Conversations
Use threads to maintain context in interactive conversations:
```python theme={null}
import marvin
with marvin.Thread() as thread:
# Get user preferences
preferences = marvin.run(
"Ask about the user's movie preferences",
cli=True
)
# Make a recommendation
recommendation = marvin.run(
"Recommend a movie based on their preferences"
)
# Ask for feedback
feedback = marvin.run(
"Ask if they'd like another recommendation",
cli=True
)
```
## Example: Interactive Survey
Here's a complete example of an interactive survey:
```python Code theme={null}
import marvin
from typing import Literal
from dataclasses import dataclass
@dataclass
class Survey:
name: str
age: int
interests: list[str]
experience_level: Literal["beginner", "intermediate", "advanced"]
with marvin.Thread() as thread:
# Create an interactive agent
agent = marvin.Agent(
name="Surveyor",
cli=True,
instructions="Conduct a friendly survey about programming experience."
)
# Run the survey
result = marvin.run(
"Conduct a programming survey",
agents=[agent],
result_type=Survey
)
print(result)
```
```text Result theme={null}
Agent: Hi! I'd like to ask you a few questions about your programming experience.
What's your name?
User: Alice
Agent: Nice to meet you, Alice! Could you tell me your age?
User: 25
Agent: Thanks! What programming languages or technologies are you interested in?
Please list them separated by commas.
User: Python, JavaScript, Machine Learning
Agent: Great interests! How would you rate your overall programming experience level?
Choose one: beginner, intermediate, or advanced
User: intermediate
Survey(
name='Alice',
age=25,
interests=['Python', 'JavaScript', 'Machine Learning'],
experience_level='intermediate'
)
```
# Memory
Source: https://askmarvin.ai/patterns/memory
Give your agents persistent memory across conversations.
While threads maintain conversation history within a single interaction, Marvin's memory feature allows agents to store and recall information across multiple conversations. This is useful for:
* Remembering user preferences across sessions
* Building up knowledge bases over time
* Sharing information between different agents
* Maintaining context across multiple interactions
* Creating more personalized user experiences
## How Memory Works
Marvin memories are implemented as vector stores that allow agents to store and retrieve information using natural language. Each memory has a unique `key` that identifies it and instructions that guide how it should be used.
## Creating Memory Modules
To create a memory module, you need to provide a `key` and `instructions`:
```python theme={null}
import marvin
# Create a memory for user preferences
preferences = marvin.Memory(
key="user_preferences",
instructions="Store and retrieve user preferences and settings."
)
```
### Memory Keys
The `key` uniquely identifies a memory module and must be consistent across uses to access the same stored information. Choose descriptive, unique keys for different types of information:
```python theme={null}
# Different memories for different purposes
user_prefs = marvin.Memory(key="user_preferences")
knowledge_base = marvin.Memory(key="documentation")
project_info = marvin.Memory(key="project_details")
```
### Memory Instructions
The `instructions` tell agents how to use the memory - what to store, when to access it, and how to format information:
```python theme={null}
weather_memory = marvin.Memory(
key="weather",
instructions="""
Store daily weather information including:
- Temperature
- Conditions
- Location
Format: 'Location: temp, conditions'
"""
)
```
## Using Memories
### With Tasks
Attach memories to tasks to provide context:
```python theme={null}
import marvin
# Create a memory module
weather = marvin.Memory(key="weather")
# Use it in a task
task = marvin.Task(
instructions="What was yesterday's weather?",
memories=[weather]
)
```
### With Agents
Give agents persistent memory across all their tasks:
```python theme={null}
import marvin
# Create an agent with memory
agent = marvin.Agent(
name="WeatherBot",
memories=[weather]
)
# The agent can access weather memory in any task
marvin.run("Record today's weather", agents=[agent])
marvin.run("Compare to yesterday's weather", agents=[agent])
```
## Example: Weather Tracking
Here's a complete example of using memory to track weather information:
```python theme={null}
import marvin
# Create a memory module
weather = marvin.Memory(
key="weather",
instructions="Store and retrieve daily weather information."
)
# Record today's weather
marvin.run(
"Record that it's 72°F and sunny today",
memories=[weather]
)
# Later, retrieve the information
result = marvin.run(
"What was the weather like today?",
memories=[weather]
)
print(result)
```
```text theme={null}
Today's weather was 72°F and sunny.
```
## Memory Providers
Marvin supports several vector store backends for memory storage:
### ChromaDB (Default)
ChromaDB is the default provider, offering a simple, file-based vector store:
```bash pip theme={null}
pip install "marvin[chroma]"
```
```bash uv theme={null}
uv add "marvin[chroma]"
```
Configure Chroma:
```python theme={null}
import marvin
from marvin.memory.providers import chroma
provider = chroma.ChromaEphemeralMemory()
# or provider = chroma.ChromaPersistentMemory()
# or provider = chroma.ChromaCloudMemory()
memory = marvin.Memory(
key="knowledge",
provider=provider
)
```
### LanceDB
LanceDB provides a fast, efficient vector store with columnar storage:
```bash pip theme={null}
pip install "marvin[lance]"
```
```bash uv theme={null}
uv add "marvin[lance]"
```
Configure LanceDB:
```python theme={null}
import marvin
from marvin.memory.providers import LanceProvider
provider = LanceProvider(uri="data/memories.lance")
memory = marvin.Memory(
key="knowledge",
provider=provider
)
```
### PostgreSQL
Use PostgreSQL with pgvector for scalable, production-ready vector storage:
```bash pip theme={null}
pip install "marvin[postgres]"
```
```bash uv theme={null}
uv add "marvin[postgres]"
```
Configure PostgreSQL:
```python theme={null}
import marvin
from marvin.memory.providers import PostgresProvider
provider = PostgresProvider(
connection_string="postgresql://user:pass@localhost/db"
)
memory = marvin.Memory(
key="knowledge",
provider=provider
)
```
### Qdrant
[Qdrant](https://qdrant.tech/) is a high-performance vector search engine built with Rust.
```bash pip theme={null}
pip install "marvin" "qdrant-client[fastembed]"
```
```bash uv theme={null}
uv add marvin "qdrant-client[fastembed]"
```
Configure Qdrant:
```python theme={null}
import marvin
from marvin.memory.providers import QdrantProvider
from qdrant_client import AsyncQdrantClient
provider = QdrantProvider(client=AsyncQdrantClient())
memory = marvin.Memory(
key="knowledge",
provider=provider
)
```
### All Providers
Install all memory providers at once:
```bash pip theme={null}
pip install "marvin[memory]" # Includes chroma, lance, postgres, qdrant
```
```bash uv theme={null}
uv add "marvin[memory]" # Includes chroma, lance, postgres and qdrant
```
# Running Tasks
Source: https://askmarvin.ai/patterns/running-tasks
Execute tasks and manage their results.
Tasks represent discrete units of work that need to be completed by AI agents. Marvin provides several ways to execute tasks and retrieve their results.
## Creating and Running Tasks
The most convenient way to run a task is with the `marvin.run()` function. This function creates and runs a task in a single call, accepting all the same arguments as the `Task` constructor:
```python Code theme={null}
import marvin
poem = marvin.run("Write a poem about AI")
print(poem)
```
```text Result theme={null}
In circuits deep and code profound,
An AI's mind begins to sound.
Electric thoughts and data streams,
Crafting worlds and shaping dreams.
```
### Running a Single Task
You can also create a task first and run it later using the `run()` method:
```python Code theme={null}
import marvin
task = marvin.Task("Write a poem about AI")
poem = task.run()
print(poem)
```
```text Result theme={null}
In circuits deep and code profound,
An AI's mind begins to sound.
Electric thoughts and data streams,
Crafting worlds and shaping dreams.
```
### Running Multiple Tasks
To run multiple tasks at once, use the `run_tasks()` function:
```python theme={null}
import marvin
task_1 = marvin.Task("Write a poem about AI")
task_2 = marvin.Task("Critique the poem", depends_on=[task_1])
results = marvin.run_tasks([task_1, task_2])
# Can also access task_1.result and task_2.result
```
## Task Dependencies
Tasks can depend on other tasks, ensuring they run in the correct order:
```python theme={null}
import marvin
name_task = marvin.Task("Get the user's name", cli=True)
poem_task = marvin.Task(
"Write a poem about the user",
depends_on=[name_task]
)
# Running poem_task will automatically run name_task first
poem = poem_task.run()
```
When you run a task, Marvin will:
1. Check if there are any dependencies
2. Run any incomplete dependencies first
3. Execute the task itself
4. Return the result
## Task Results
Tasks can specify their expected result type using the `result_type` parameter:
```python theme={null}
import marvin
# Get a boolean result
is_spam = marvin.run(
"Is this email spam?",
result_type=bool,
context={"email": "You just won a million dollars!"}
)
# Get a list of integers
rolls = marvin.run(
"Roll 3 dice",
result_type=list[int],
tools=[roll_dice]
)
```
## Task Status
You can check a task's status at any time:
```python theme={null}
import marvin
task = marvin.Task("Write a poem")
# Before running
print(task.status) # TaskStatus.PENDING
# After running
task.run()
print(task.status) # TaskStatus.SUCCESSFUL
```
Tasks have several helper methods for checking their status:
* `is_pending()`: Task hasn't started
* `is_running()`: Task is currently executing
* `is_successful()`: Task completed successfully
* `is_failed()`: Task failed to complete
* `is_skipped()`: Task was skipped
* `is_complete()`: Task is done (successful, failed, or skipped)
* `is_incomplete()`: Task isn't done (pending or running)
* `is_ready()`: Task can be run (dependencies are complete)
## Task Context
Tasks maintain their own context, which can include:
* Instructions: What needs to be done
* Context: Additional information needed for the task
* Tools: Functions the agent can use
* Memories: Persistent information from previous runs
* Result type: Expected format of the output
This context helps agents understand and complete the task effectively:
```python theme={null}
import marvin
task = marvin.Task(
instructions="Analyze the sentiment of this text",
context={"text": "I love this product!"},
result_type=str,
memories=[knowledge_base],
tools=[analyze_text]
)
```
# Task Results
Source: https://askmarvin.ai/patterns/task-results
Get structured outputs from your tasks.
Marvin provides powerful ways to control and validate the output of tasks. By specifying a `result_type`, you can ensure that tasks return exactly the data structure you need.
## Scalar Types
The simplest way to specify a result type is with Python's built-in scalar types:
```python theme={null}
import marvin
# Get a string
text = marvin.run(
"Write a haiku",
result_type=str
)
# Get a number
temperature = marvin.run(
"Convert 72°F to Celsius",
result_type=float
)
# Get a boolean
is_spam = marvin.run(
"Is this email spam?",
result_type=bool,
context={"email": "You just won a million dollars!"}
)
```
## Classification
For classification tasks, Marvin provides several ways to specify options. Under the hood, Marvin optimizes classification by having agents choose indices rather than writing out full labels:
```python theme={null}
import marvin
from enum import Enum
from typing import Literal
# Using a list of values (most flexible)
sentiment = marvin.run(
"Classify the sentiment of this review",
result_type=["positive", "negative", "neutral"],
context={"text": "This product exceeded my expectations!"}
)
print(sentiment) # "positive"
# Using Literal type
status = marvin.run(
"Check the server status",
result_type=Literal["up", "down", "maintenance"]
)
# Using an Enum
class Sentiment(Enum):
POSITIVE = "positive"
NEGATIVE = "negative"
NEUTRAL = "neutral"
sentiment = marvin.run(
"Classify the sentiment",
result_type=Sentiment
)
print(sentiment) # Sentiment.POSITIVE
print(sentiment.value) # "positive"
```
### Multi-label Classification
For tasks where multiple labels can apply, use either `list[Literal]`, `list[Enum]`, or Marvin's shorthand double-list syntax:
```python theme={null}
# Using list[Literal]
topics = marvin.run(
"What topics does this article cover?",
result_type=list[Literal["technology", "science", "business", "politics"]],
context={"text": "AI startups are revolutionizing healthcare..."}
)
print(topics) # ["technology", "science", "business"]
# Using list[Enum]
topics = marvin.run(
"What topics apply?",
result_type=list[Sentiment]
)
print([t.value for t in topics]) # ["positive", "neutral"]
# Using shorthand double-list syntax
topics = marvin.run(
"What topics does this article cover?",
result_type=[["technology", "science", "business", "politics"]],
context={"text": "AI startups are revolutionizing healthcare..."}
)
print(topics) # ["technology", "science", "business"]
```
## Collections
Use Python's type hints to specify collections. Marvin supports several collection types:
### Lists
Lists are the most common collection type and work well with all LLM providers:
```python theme={null}
import marvin
# Get a list of strings
keywords = marvin.run(
"Extract keywords from this text",
result_type=list[str],
context={"text": "AI and machine learning are transforming..."}
)
# Get a list of numbers
prices = marvin.run(
"Extract all prices from this text",
result_type=list[float],
context={"text": "The shirt costs $19.99 and the pants are $49.99"}
)
```
### Sets
Sets work similarly to lists but ensure unique values:
```python theme={null}
import marvin
# Get unique words
unique_words = marvin.run(
"What unique words appear in this text?",
result_type=set[str],
context={"text": "the cat and the dog"}
)
# Get unique numbers
numbers = marvin.run(
"Give me 5 unique numbers between 1 and 10",
result_type=set[int]
)
```
### Dictionaries
Dictionaries are useful for key-value data:
```python theme={null}
import marvin
# Get a simple word count
word_counts = marvin.run(
"Count word frequencies",
result_type=dict[str, int],
context={"text": "the cat and the dog"}
)
# Get nested data
character_details = marvin.run(
"Describe these characters' ages",
result_type=dict[str, dict[str, int]],
context={"characters": ["Luke", "Leia", "Han"]}
)
# Mix with other types
movie_ratings = marvin.run(
"Rate these movies from 1-5",
result_type=dict[str, float],
context={"movies": ["The Matrix", "Inception"]}
)
```
### Tuples
While tuples are not directly supported by most LLM providers, Marvin will attempt to coerce the result into a tuple for you:
```python theme={null}
import marvin
# Get coordinates
coordinates = marvin.run(
"Convert '40.7128° N, 74.0060° W' to decimal coordinates",
result_type=tuple[float, float]
)
# Get name and age
person_info = marvin.run(
"Extract name and age from: John is 25 years old",
result_type=tuple[str, int]
)
```
## Structured Types
Marvin supports several options for complex data structures, each with their own benefits:
### TypedDict
TypedDicts provide a way to specify dictionary types with fixed keys:
```python theme={null}
import marvin
from typing import TypedDict
class MovieDict(TypedDict):
title: str
year: int
rating: float
movie = marvin.run(
"Describe the movie 'Inception'",
result_type=MovieDict
)
print(movie["title"]) # "Inception"
```
### Dataclasses
Dataclasses offer a more object-oriented approach with attribute access:
```python theme={null}
import marvin
from dataclasses import dataclass
@dataclass
class Movie:
title: str
year: int
rating: float
movie = marvin.run(
"Describe the movie 'Inception'",
result_type=Movie
)
print(movie.title) # "Inception"
```
### Pydantic Models
Pydantic models (recommended) provide rich validation and nested structures:
```python theme={null}
import marvin
from dataclasses import dataclass
@dataclass
class Person:
name: str
age: int
interests: list[str]
@dataclass
class Movie:
title: str
director: Person
year: int
genres: list[str]
rating: float
# Get structured movie data
movie = marvin.run(
"Describe the movie 'Inception'",
result_type=Movie
)
print(movie.director.name) # "Christopher Nolan"
# Get a list of people
people = marvin.run(
"List the main characters in Star Wars",
result_type=list[Person]
)
print(people[0].name) # "Luke Skywalker"
```
## Validation
### Result Validators
You can provide a validation function to enforce additional constraints. The function should either return the validated result or raise an exception:
```python theme={null}
import marvin
def validate_even(value: int) -> int:
if value % 2 != 0:
print(f"tried {value!r}")
raise ValueError("Value must be even")
return value
number = marvin.run(
"Give me a number close to 42",
result_type=int,
result_validator=validate_even
)
print(number)
```
```python theme={null}
tried 41
44
```
### Pydantic Validation
When using Pydantic models, you can use field or model validators for more complex validation:
#### using `field_validator`
```python theme={null}
from pydantic import BaseModel, field_validator
import marvin
class User(BaseModel):
username: str
password: str
@field_validator("password")
def validate_password(cls, v):
if len(v) < 8 or v.lower() == v:
print(f"tried {v!r}")
raise ValueError("must be >= 8 characters and not all lowercase")
return v
user = marvin.run(
"Gilfoyle joined the company and asked for password: 'dineshsux'",
result_type=User
)
print(user)
```
```python theme={null}
tried 'dineshsux'
username='Gilfoyle' password='DineshSux123'
```
#### using `model_validator`
```python theme={null}
from pydantic import BaseModel, model_validator
import marvin
class User(BaseModel):
username: str
password: str
is_admin: bool = False
@model_validator(mode="after")
def validate_user(self):
if self.username == "BigHead":
print("let's randomly make him admin")
self.is_admin = True
return self
user = marvin.run(
"idk how, but BigHead joined the company",
result_type=User
)
print(user)
```
```python theme={null}
let's randomly make him admin
username='BigHead' password='abc123ABC' is_admin=True
```
### Annotated Types
Use `Annotated` to provide additional context about the expected result:
```python theme={null}
from pydantic import Field
from typing import Annotated
import marvin
# Get a specific format
print(
marvin.run(
"What's the zip code for Manhattan?",
result_type=Annotated[str, Field(max_length=5)]
)
)
# Get a constrained number
print(
marvin.run(
"Rate this movie",
result_type=Annotated[int, Field(gt=0, lte=5)]
)
)
```
```python theme={null}
10001
4
```
If you want to write a function (like `field_validator`) to validate a type so that the validation is baked into the type itself, you can use `pydantic`'s functional validators:
```python theme={null}
from typing import Annotated
from pydantic import BeforeValidator
import marvin
def ensure_random_number(v):
if v != 3141592653:
print(f"tried {v}")
raise ValueError("everyone knows the most random number is 3141592653!")
return v
MostRandomNumber = Annotated[int, BeforeValidator(ensure_random_number)]
result = marvin.run(
"What is the most random number?",
result_type=MostRandomNumber
)
print(result)
```
```python theme={null}
tried 42
3141592653
```
# Tools
Source: https://askmarvin.ai/patterns/tools
Give agents new abilities with custom tools.
A tool is a Python function that your agents can use to accomplish tasks. Tools extend an agent's capabilities beyond their built-in knowledge and abilities, allowing them to interact with external systems, perform calculations, or access specific information.
Tools can be:
* Simple utility functions for common operations
* Complex data processing operations
* API calls to external services
* File system operations
* Database queries
* System commands
* Custom business logic
Here's a basic example of a tool for rolling dice:
```python theme={null}
import marvin
import random
def roll_die() -> int:
"""Roll a 6-sided die."""
return random.randint(1, 6)
rolls = marvin.run("Roll a die four times", tools=[roll_die])
print(rolls)
```
```text Result theme={null}
[3, 1, 4, 2]
```
## Creating Tools
Any Python function can be used as a tool. The function's name, docstring, and type hints help agents understand what the tool does and how to use it:
```python theme={null}
from typing import Annotated
from pydantic import Field
Location = Annotated[str, Field(description="The 5 or 9-digit zip code")]
Temperature = Annotated[float, Field(description="The temperature in Fahrenheit")]
def get_weather(location: Location) -> Temperature:
"""Fetch weather data for the specified location."""
# Implementation details...
```
## Using Tools
Tools can be provided to tasks or agents. When a tool is provided to a task, any agent working on that task will have access to the tool. When a tool is provided to an agent, the agent can use the tool in any task it is assigned to:
```python theme={null}
import marvin
def search_database(query: str) -> list:
"""Search the product database for items matching the query."""
# Implementation details...
# Provide tools to a task
task = marvin.Task(
instructions="Find products matching 'blue shirt'",
tools=[search_database]
)
# Or provide tools to an agent
agent = marvin.Agent(
name="ProductBot",
tools=[search_database]
)
```
## Common Use Cases
Tools are particularly useful for:
* Accessing external APIs or services
* Performing calculations or data processing
* File operations and data manipulation
* Database queries and updates
* System interactions
* Custom business logic
* Integration with existing code
# Quickstart
Source: https://askmarvin.ai/quickstart
Build your first AI application in less than a minute
Welcome to Marvin! This quickstart guide will walk you through the three main ways to use Marvin to create AI-powered applications.
You'll learn how to:
1. [Use quick one-liners](#quick-one-liners) with `marvin.run()`
2. [Work with specialized agents](#specialized-agents) using `agent.run()`
3. [Control tasks directly](#task-control) with `task.run()`
4. [Add advanced features](#advanced-features) like context and threads
This guide assumes you have already installed `marvin`. See the [installation docs](/installation) for instructions.
## Quick One-Liners
The fastest way to use Marvin is with `marvin.run()`:
```python theme={null}
import marvin
# Run a simple task
print(marvin.run("Write a haiku about Python programming"))
# With a specific result type
numbers = marvin.run(
"Generate five random numbers between 1 and 10",
result_type=list[int]
)
```
```text theme={null}
Indented lines flow
Functions dance with elegance
Code becomes poetry
[3, 7, 2, 9, 5]
```
## Specialized Agents
Create and use specialized agents for specific types of tasks:
```python theme={null}
from marvin import Agent
# Create a specialized agent
writer = Agent(
name="Technical Writer",
instructions="Write clear, engaging content for developers"
)
# Use the agent directly
article = writer.run(
"Write a short article about Python type hints",
result_type=str
)
print(article)
```
```text theme={null}
Type Hints in Python: A Clear Path to Better Code
Python's type hints bring clarity and safety to your code without sacrificing its dynamic nature. Introduced in Python 3.5, type hints allow developers to explicitly declare variable and function types, making code more maintainable and easier to understand.
Here's a quick example:
def greet(name: str) -> str:
return f"Hello, {name}!"
Type hints help catch errors early, improve IDE support, and make your code self-documenting. While optional, they're becoming an essential tool in modern Python development.
```
## Task Control
For full control over your AI workflows, create and run tasks directly:
```python theme={null}
import httpx
from marvin import Task
from pydantic import BaseModel
def check_weather(location: str) -> str:
url = f"https://wttr.in/{location}?format=%C+%t"
return httpx.get(url).text
# Define a structured output type
class WeatherForecast(BaseModel):
temperature: float
conditions: str
precipitation_chance: float | str = "unknown"
# Create a task with specific requirements
task = Task(
instructions="Get the weather for San Francisco",
result_type=WeatherForecast,
tools=[check_weather] # Custom tools
)
# Run the task
forecast = task.run()
print(forecast)
```
```text theme={null}
temperature=50.0 conditions='Partly cloudy' precipitation_chance='unknown'
```
## Advanced Features
All three approaches support advanced features like context, threads, and tools:
```python theme={null}
import marvin
from marvin import Agent, Task
# Context improves results
data = "The patient reports mild fever and fatigue."
diagnosis = marvin.run(
"Suggest possible conditions",
context={"medical_notes": data}
)
# Threads maintain conversation history
with marvin.Thread() as thread:
# Ask multiple related questions
marvin.run("What is quantum computing?")
marvin.run("How does that relate to classical computing?")
marvin.run("What are its practical applications?")
# Teams of agents can collaborate
researcher = Agent("Researcher")
writer = Agent("Technical Writer")
editor = Agent("Editor")
with marvin.Thread() as thread:
research = researcher.run("Research quantum computing")
draft = writer.run("Write an article", context={"research": research})
final = editor.run("Edit the article", context={"draft": draft})
```
## What's Next?
You've seen the main ways to work with Marvin. To learn more:
* Read about [Tasks](/concepts/tasks) to understand how to structure AI workflows
* Learn about [Agents](/concepts/agents) to create specialized AI workers
* Explore [Threads](/concepts/threads) for managing conversation context
* Check out [Models](/concepts/models) to work with structured data
# Marvin
Source: https://askmarvin.ai/welcome
A powerful framework for building AI applications
![]()
## What is Marvin?
**Marvin is a Python framework for building AI applications with LLMs.**
Marvin provides a clean, intuitive interface for working with large language models (LLMs) while handling all the complexity of state management, conversation history, and agent coordination.
## Quick Examples
Using `uv`? Run the examples below by copying them to your clipboard and executing:
```bash macOS theme={null}
pbpaste | uv run --with marvin -
```
```bash linux theme={null}
xclip -selection clipboard -o | uv run --with marvin -
```
```bash windows theme={null}
Get-Clipboard | uv run --with marvin -
```
### Simple Tasks
The fastest way to use Marvin is with `marvin.run()`:
```python Run a simple task theme={null}
import marvin
print(marvin.run("Write a haiku about artificial intelligence"))
```
```python Provide context for more accurate results theme={null}
import httpx
import marvin
print(
marvin.run(
"Summarize the current astronauts in space like a 50s news anchor",
context={"data": httpx.get("http://api.open-notify.org/astros.json", headers={"Accept": "application/json"}).json()}
)
)
```
### Specialized Agents
Create agents with specific skills and personalities:
```python theme={null}
import marvin
# Create specialized agents
poet = marvin.Agent(
name="Poet",
instructions="You are an expert poet who writes in various styles."
)
scientist = marvin.Agent(
name="Scientist",
instructions="You are a scientist who explains complex topics clearly."
)
# Use them for specific tasks
explanation = marvin.run(
"Explain entropy briefly",
agents=[scientist]
)
poem = marvin.run(
"Write a haiku about entropy",
agents=[poet],
context={"scientific_background": explanation}
)
print(poem)
```
### Structured Data
Get results in exactly the format you need:
```python theme={null}
from typing import Annotated, Literal
import marvin
from pydantic import BaseModel, Field
class Character(BaseModel):
name: str
role: Literal["hero", "villain", "sidekick"]
aura: Annotated[float, Field(ge=0, le=1)]
# Get structured results
characters = marvin.run(
"Create three characters for a mystery story",
result_type=list[Character]
)
for char in characters:
print(f"{char.name} - {char.role} - {char.aura}")
```
### Persistent Memory
Give your agents memory that persists across conversations:
```python theme={null}
import marvin
# Create a memory module
preferences = marvin.Memory(
key="user_preferences",
instructions="Remember user preferences and style"
)
# Create an agent with memory
assistant = marvin.Agent(memories=[preferences])
# The agent will remember information across conversations
marvin.run(
"Learn about the user's writing style preferences",
agents=[assistant],
cli=True
)
```
## Why Marvin?
* 🎯 **Simple Interface**: Start with one line of code, scale to complex applications
* 🧠 **Smart Defaults**: Sensible defaults that just work, with customization when you need it
* 🔄 **State Management**: Built-in conversation history and memory management
* 📝 **Structured Data**: Get results in exactly the format you need
* 🤝 **Multi-Agent**: Create specialized agents that work together
* 🔌 **Extensible**: Easy to integrate with your existing Python code
## Next Steps
* [Install Marvin](/installation) - Get started in under a minute
* Try the [Quickstart](/quickstart) - Build your first AI application
* Explore [Core Concepts](/concepts) - Learn about tasks, agents, and more
* Browse [Examples](https://github.com/prefecthq/marvin/tree/main/examples) - See Marvin in action