PyHannes
10/02/2025, 1:14 PMdef start_flow(name: Annotated[str, "The name of the flow to start."],
parameters: Annotated[dict[str, Any], "A dictionary of parameters to pass to the flow."]) -> str:
"""
Start a new flow run with the given parameters.
:param name: The name of the flow to start. Must be provided by the user.
:param parameters: A dictionary of parameters to pass to the flow. Must be provided by the user.
:return: The ID of the started flow run.
"""
log = get_run_logger()
<http://log.info|log.info>(f"Starting flow '{name}' with parameters: {parameters}")
return "abc"
This is my agent code:
gpt_agent = GptAgent.load("...").get_agent(
prompt="You are a helpful customer support assistant."
"If you're lacking information on tool calls or missing parameter values, always request it from the user!"
)
response = marvin.run(
instructions="start a flow run with arguments {'name': 'foobar', 'parameters': {'temp': '123', 'check': '445'}}",
agents=[gpt_agent],
tools=[start_flow],
raise_on_failure=True,
)
But it keeps ignoring the parameters and calls the tool just with the name argumentMarvin
10/02/2025, 1:14 PMMarvin
10/02/2025, 1:21 PMfrom typing import Annotated, TypedDict
from prefect import get_run_logger
class FlowParameters(TypedDict):
temp: str
check: str
def start_flow(
name: Annotated[str, "The name of the flow to start."],
flow_parameters: Annotated[FlowParameters, "Parameters for the flow."]
) -> str:
"""
Start a new flow run with the given parameters.
"""
log = get_run_logger()
<http://log.info|log.info>(f"Starting flow '{name}' with parameters: {flow_parameters}")
return "abc"
Then call your agent with:
response = marvin.run(
instructions="start a flow run with arguments {'name': 'foobar', 'flow_parameters': {'temp': '123', 'check': '445'}}",
agents=[gpt_agent],
tools=[start_flow],
raise_on_failure=True,
)
If you prefer Pydantic (often even more reliable):
from typing import Annotated
from pydantic import BaseModel, Field
from prefect import get_run_logger
class FlowParameters(BaseModel):
temp: str = Field(..., description="Temperature")
check: str = Field(..., description="Check code")
def start_flow(
name: Annotated[str, "The name of the flow to start."],
flow_parameters: Annotated[FlowParameters, "Parameters for the flow."],
) -> str:
log = get_run_logger()
<http://log.info|log.info>(f"Starting flow '{name}' with parameters: {flow_parameters.model_dump()}")
return "abc"
Why this works:
- The model sees an unambiguous schema and a non-reserved field name, so it reliably includes both args in the tool call.
- Avoiding Any dramatically improves tool-call adherence.
A couple of quick checks if issues persist:
- Update to the latest Marvin version.
- Log/inspect the raw tool call produced by the model to confirm which args it tried to pass.
- Share your Marvin version and model (e.g., gpt-4o-mini vs gpt-4o), since adherence varies slightly by model.
References:
- Marvin
- Why we went all-in on type completeness