Add support for Free-Form Function Calling and Context Free Grammar constraints over string output

[matthewfranglen]

Description

The release of GPT-5 has added a different form of structured output. This uses a context free grammar implemented using either the LARK or regex formats. It is described here.

With this you can write agents that can reliably produce structured output strings. The example in the link above is for SQL, I have also found this useful for the lucene query language. Composing valid queries of arbitrary complexity using pydantic models is both tricky programmatically and very likely difficult for a given LLM to work with.

I would like to propose the addition of an output_grammar to the agent constructor and or run method which takes a suitable grammar, validates it, and then provides it to the completion endpoint. It would be an error to provide the grammar if the output_type is not a string.

References

https://cookbook.openai.com/examples/gpt-5/gpt-5_new_params_and_tools#3-contextfree-grammar-cfg

my lucene experiment:

from openai import OpenAI
from lark import Lark

client = OpenAI()

prompt = (
    "Call the lucene_query to generate a query for the Solr database that will "
    "match the documents that reference coca-cola by name or alias"
)
grammar = LUCENE_GRAMMAR_FILE.read_text()

# check validity of grammar
Lark(grammar, parser="lalr")

# make request, this works with gpt-5 and gpt-5-mini
response = client.responses.create(
    model="gpt-5",
    input=prompt,
    text={"format": {"type": "text"}},
    tools=[
        {
            "type": "custom",
            "name": "lucene_query",
            "description": (
                "Executes read-only lucene queries that match the text of "
                "the document. YOU MUST REASON HEAVILY ABOUT THE QUERY AND "
                "MAKE SURE IT OBEYS THE GRAMMAR."
            ),
            "format": {
                "type": "grammar",
                "syntax": "lark",
                "definition": grammar,
            },
        },
    ],
    parallel_tool_calls=False,
)

print("--- Lucene Query ---")
print(response.output[1].input)

The grammar is available here.

[matthewfranglen]

I've been working on this today and I came up with the following custom OpenAI model:

from typing import Literal, cast

from lark import Lark
from openai import NOT_GIVEN, APIStatusError, AsyncOpenAI
from openai.types import responses
from openai.types.responses import CustomToolParam
from pydantic_ai import ModelHTTPError
from pydantic_ai._utils import (
    number_to_datetime,
)
from pydantic_ai.messages import (
    ModelMessage,
    ModelRequest,
    ModelResponse,
    ModelResponsePart,
    TextPart,
    ThinkingPart,
)
from pydantic_ai.models import (
    Model,
    ModelRequestParameters,
    check_allow_model_requests,
    get_user_agent,
)
from pydantic_ai.models.openai import (
    OpenAIModelName,
    OpenAIResponsesModel,
    OpenAIResponsesModelSettings,
    _map_usage,
)
from pydantic_ai.profiles import ModelProfileSpec
from pydantic_ai.providers import Provider
from pydantic_ai.settings import ModelSettings


class CFGOpenAIModel(Model):
    """A model that uses OpenAI's Responses API with Context-Free Grammar constraints.

    This model provides CFG-constrained generation by using OpenAI's custom tools with
    grammar format specifications.
    """

    def __init__(
        self,
        model_name: OpenAIModelName,
        *,
        provider: (
            Literal[
                "openai",
                "deepseek",
                "azure",
                "openrouter",
                "grok",
                "fireworks",
                "together",
            ]
            | Provider[AsyncOpenAI]
        ) = "openai",
        profile: ModelProfileSpec | None = None,
        settings: ModelSettings | None = None,
        grammar: str,
        grammar_syntax: Literal["lark", "regex"] = "lark",
        tool_name: str = "cfg_tool",
        tool_description: str = "Generate output conforming to the specified grammar",
    ) -> None:
        self._model_name = model_name
        self._grammar = grammar
        self._grammar_syntax: Literal["lark", "regex"] = grammar_syntax
        self._tool_name = tool_name
        self._tool_description = tool_description

        # Validate grammar
        if grammar_syntax == "lark":
            try:
                Lark(grammar)
            except Exception as e:
                raise ValueError(f"Invalid Lark grammar: {e}") from e

        # Create a regular OpenAI model as fallback for properties
        self.model = OpenAIResponsesModel(
            model_name=model_name,
            provider=provider,
            profile=profile,
            settings=settings,
        )

    @property
    def model_name(self) -> OpenAIModelName:
        """The model name."""
        return self._model_name

    @property
    def system(self) -> str:
        """The system identifier."""
        return "openai"

    async def request(
        self,
        messages: list[ModelMessage],
        model_settings: ModelSettings | None,
        model_request_parameters: ModelRequestParameters,
    ) -> ModelResponse:
        """Make a CFG-constrained request using OpenAI Responses API."""
        check_allow_model_requests()
        response = await self._responses_create(
            messages,
            cast("OpenAIResponsesModelSettings", model_settings or {}),
            model_request_parameters,
        )
        return self._process_response(response)

    async def _responses_create(
        self,
        messages: list[ModelRequest | ModelResponse],
        model_settings: OpenAIResponsesModelSettings,
        model_request_parameters: ModelRequestParameters,
    ) -> responses.Response:
        check_allow_model_requests()

        if (
            model_request_parameters.builtin_tools
            or model_request_parameters.function_tools
            or model_request_parameters.output_tools
        ):
            raise ValueError(
                "CFG models cannot use additional tools - grammar constrains entire output",
            )

        instructions, openai_messages = await self.model._map_messages(messages)
        reasoning = self.model._get_reasoning(model_settings)

        max_tokens = model_settings.get("max_tokens", 5000)
        tool_definition: CustomToolParam = {
            "type": "custom",
            "name": self._tool_name,
            "description": self._tool_description,
            "format": {
                "type": "grammar",
                "syntax": self._grammar_syntax,
                "definition": self._grammar,
            },
        }

        try:
            extra_headers = model_settings.get("extra_headers", {})
            extra_headers.setdefault("User-Agent", get_user_agent())
            return await self.model.client.responses.create(
                input=openai_messages,
                model=self._model_name,
                instructions=instructions,
                parallel_tool_calls=False,
                tools=[tool_definition],
                tool_choice={"name": self._tool_name, "type": "custom"},
                max_output_tokens=max_tokens,
                stream=False,
                truncation="disabled",
                timeout=model_settings.get("timeout", NOT_GIVEN),
                service_tier=model_settings.get("openai_service_tier", NOT_GIVEN),
                reasoning=reasoning,
                user=model_settings.get("openai_user", NOT_GIVEN),
                text={"format": {"type": "text"}},
                extra_headers=extra_headers,
                extra_body=model_settings.get("extra_body"),
            )
        except APIStatusError as e:
            if (status_code := e.status_code) >= 400:
                raise ModelHTTPError(
                    status_code=status_code,
                    model_name=self.model_name,
                    body=e.body,
                ) from e
            raise  # pragma: lax no cover

    def _process_response(self, response: responses.Response) -> ModelResponse:
        """Process a non-streamed response, and prepare a message to return."""
        timestamp = number_to_datetime(response.created_at)
        items: list[ModelResponsePart] = []
        for item in response.output:
            if item.type == "reasoning":
                for summary in item.summary:
                    # NOTE: We use the same id for all summaries because we can merge them on the round trip.
                    # The providers don't force the signature to be unique.
                    items.append(ThinkingPart(content=summary.text, id=item.id))
            elif item.type == "custom_tool_call":
                items.append(TextPart(item.input))
        return ModelResponse(
            items,
            usage=_map_usage(response),
            model_name=response.model,
            vendor_id=response.id,
            timestamp=timestamp,
        )

It's heavily based on the original and can be run like so:

system_prompt = _load_system_prompt()
tool_description = _load_tool_description()
grammar = _load_grammar()
agent = Agent(
    model=CFGOpenAIModel(
        model_name="gpt-5-mini",
        grammar=grammar,
        grammar_syntax="lark",
        tool_name="lucene_query",
        tool_description=tool_description,
        settings={
            "max_tokens": 5000,
            "openai_reasoning_effort": "low",
        },
    ),
    system_prompt=system_prompt,
    output_type=str,
)

the token count and reasoning effort are correlated, if the reasoning is too high then no valid output will be generated.

I can turn this into a PR if it seems appropriate?

[DouweM]

@matthewfranglen That's a nice feature that'd be awesome to support natively!

CFG support builds on Free-Form Function Calling (FFFC) described in that same doc, so I'm thinking we should support this both for regular tools that take a single str argument, as well as output tools for the str type.

For FFFC, we just need to add "type": "custom" to the tool definition sent to the API. That'd be useful for tools that take Python code, for example.

For CFG, we need that, plus:

"format": {
    "type": "grammar",
    "syntax": "regex" | "lark",
    "definition": re.Pattern | str
}

In Pydantic AI, this would mean new arguments on @agent.tool (and @agent.tool_plain), Tool, and ToolOutput. We'd also need a new property on ModelProfile so that we can raise an error if this is passed to a model that doesn't support it.

Besides those new properties to send these fields to the API, we'd also need to parse the ResponseCustomToolCall.input into a ToolCallPart. Since that currently requires args to be a JSON string or dict, it's probably fine to wrap it in a dict with a key corresponding to the tool function str argument or 'response' in the case of ToolOutput (we're already doing something similar here and here).

I think we can capture both FFFC and CFG in a new text_format: Literal['plain'] | RegexGrammar(re.Pattern) | LarkGrammer(str) argument to the aforementioned methods and classes.

Your example would then look like output_type=ToolOutput(str, text_format=LarkGrammar(grammar)), and could work with multiple output types (tools) as well unlike the implementation you have right now. We could potentially directly support output_type=LarkGrammar(grammar) as well, but that could be added later.

Let me know what you think!

[matthewfranglen]

One part of this is the ability to enforce the grammar over the output of the agent. This is desirable for me because the lucene query can be used directly to get documents, or to get aggregates, as the situation demands. Requiring that the grammar only applies to tool calls means that all of that has to happen within the context of the individual agent call.

As I understand it the response from the agent is itself a tool call (the final_result_ tool) so if we make this a feature of tools do you think that extending that to the final tool call would be a way to return strings conforming to the grammar?

I was also thinking about this today and it would be nice to make my implementation less restricted. The agent may choose to use other tools before invoking the one that requires a schema, or desires raw input. Moving the change to the tool itself would be a neat way to achieve this.

[DouweM]

As I understand it the response from the agent is itself a tool call (the final_result_ tool) so if we make this a feature of tools do you think that extending that to the final tool call would be a way to return strings conforming to the grammar?

@matthewfranglen Yep exactly, that's what I was referring to with the ToolOutput class that would also get these new attributes: https://ai.pydantic.dev/output/#tool-output

[matthewfranglen]

Fantastic! That sounds great then. I'll have a look into the implementation more tomorrow.

[bw-matthew]

Some notes for myself mainly. If I make a mistake with these then feedback is appreciated:

The Tool can return a ToolDefinition. The openai model maps these to the tools parameter of the api call. This means that to support FFFC et al it must be possible to infer the change in signature from the ToolDefinition only.

To support direct string returns for an arbitrary function, without making any change to the Tool, then we can inspect the parameters_json_schema. This has a fixed structure for a function which takes only a string:

{"additionalProperties": false,
 "properties": {NAME: {"type": "string"}},
 "required": [NAME],
 "type": "object"}

it should be possible to match this and, for models that support it, convert the _map_tool_definition function to support this special case. A change similar to:

def _map_tool_definition(self, f: ToolDefinition) -> chat.ChatCompletionToolParam:
    tool_param: chat.ChatCompletionToolParam = {
        'type': 'function',
        'function': {
            'name': f.name,
            'description': f.description or '',
            'parameters': f.parameters_json_schema,
        },
    }
    model_profile = OpenAIModelProfile.from_profile(self.profile)
    if f.strict and model_profile.openai_supports_strict_tool_definition:
        tool_param['function']['strict'] = f.strict
    if f.only_takes_string_argument and model_profile.openai_supports_freeform_function_calling:
        tool_param["type"] = "custom"
    return tool_param

this would reduce the implementation of CFG as it can build on top of this.

EDIT: The change to the map function has to be a bit more involved than this. The chat.ChatCompletionToolParam type is not the same as that required for the custom tool call. The correct type is openai.types.responses.custom_tool_param.CustomToolParam, which has name, type, description and format keys.

[DouweM]

@matthewfranglen Unfortunately the doc states that 'custom tool type does NOT support parallel tool calling', so I don't think we should use it by default for string-only functions. I think parallel tool calls are typically more valuable than free-form tool calling (which mostly just keep the model from having to escape quotes in strings).

[matthewfranglen]

PR is here, it is WIP at the moment.