Decorator: @ivcap_lambda

The ivcap_lambda decorator registers a Python function as a set of HTTP endpoints on the FastAPI application.

ivcap_lambda(path_prefix, *, opts=ToolOptions(), context=None)

Add a few routes to the service for use with an AI tool.

The tool itself is implemented in worker_fn where the first argument is a pydantic model describing all the tool's "public" parameters. The function is also expected to return it's result as a single pydantic model. The tool function can have two optioanl paramters, one with the same type as context and the second one with fastapi.Request. The context paramter will be identical to the above context, while the request will be the incoming request.

This function then sets up three endpoints: - POST {path_prefix}: To request the execution of the tool (the 'job') - GET {path_prefix}/{job_id}: To collect the result of a tool execution - GET {path_prefix}: To obtain a description of the tool suitable for most agent frameworks

The POST request will only wait opts.max_wait_time for the tool to finish. If it hasn't finished by then, a 204 No Content code will be returned with additional header fields Location and Retry-later to inform the caller where and approx. when the result can be collected later.

The opts parameter allows for customization of the endpoints. See ToolOptions for a more detailed description.

Parameters:

Name	Type	Description	Default
path_prefix	str	The path prefix to use for this set of endpoints	required
opts	Optional[ToolOptions]	Additional behaviour settings. Defaults to ToolOptions().	ToolOptions()
context	Optional[ExecutionContext]	An optional context to be provided to every invocation of worker_fn. Defaults to None.	None

Source code in ivcap_lambda/decorators.py

def ivcap_lambda(
    path_prefix: str,
    *,
    opts: ToolOptions | None = ToolOptions(),
    context: ExecutionContext | None = None,
):
    """Add a few routes to the service for use with an AI tool.

    The tool itself is implemented in `worker_fn` where the first
    argument is a pydantic model describing all the tool's "public" parameters.
    The function is also expected to return it's result as a single pydantic model.
    The tool function can have two optioanl paramters, one with the same type as
    `context` and the second one with `fastapi.Request`. The context paramter will
    be identical to the above `context`, while the `request` will be the incoming
    request.

    This function then sets up three endpoints:
    - POST {path_prefix}: To request the execution of the tool (the 'job')
    - GET {path_prefix}/{job_id}: To collect the result of a tool execution
    - GET {path_prefix}: To obtain a description of the tool suitable for most agent frameworks

    The POST request will only wait `opts.max_wait_time` for the tool to finish. If it
    hasn't finished by then, a `204 No Content` code will be returned with additional
    header fields `Location` and `Retry-later` to inform the caller where and approx.
    when the result can be collected later.

    The `opts` parameter allows for customization of the endpoints. See `ToolOptions`
    for a more detailed description.

    Args:
        path_prefix (str): The path prefix to use for this set of endpoints
        opts (Optional[ToolOptions], optional): Additional behaviour settings. Defaults to ToolOptions().
        context (Optional[ExecutionContext], optional): An optional context to be provided to every invocation of `worker_fn`. Defaults to None.
    """

    def decorator(worker_fn: WorkerFn):
        """
        Args:
        worker_fn (Callable[[BaseModel, Optional[ExecutionContext], Optional[Response]], BaseModel]): _description_
        opts (Optional[ToolOptions], optional): Additional behaviour settings. Defaults to ToolOptions().
        context (Optional[ExecutionContext], optional): An optional context to be provided to every invocation of `worker_fn`. Defaults to None.
        """
        add_tool_api_route(
            get_fast_app(), path_prefix, worker_fn, opts=opts, context=context
        )
        return worker_fn

    return decorator

---

Related: add_tool_api_route

The lower-level function used by @ivcap_lambda when you need to register a route programmatically (e.g. in a factory pattern):

add_tool_api_route(app, path_prefix, worker_fn, *, opts=ToolOptions(), context=None)

Add a few routes to app for use with an AI tool.

The tool itself is implemented in worker_fn where the first argument is a pydantic model describing all the tool's "public" parameters. The function is also expected to return it's result as a single pydantic model. The tool function can have two optioanl paramters, one with the same type as context and the second one with fastapi.Request. The context paramter will be identical to the above context, while the request will be the incoming request.

This function then sets up three endpoints: - POST {path_prefix}: To request the execution of the tool (the 'job') - GET {path_prefix}/{job_id}: To collect the result of a tool execution - GET {path_prefix}: To obtain a description of the tool suitable for most agent frameworks

The POST request will only wait opts.max_wait_time for the tool to finish. If it hasn't finished by then, a 204 No Content code will be returned with additional header fields Location and Retry-later to inform the caller where and approx. when the result can be collected later.

The opts parameter allows for customization of the endpoints. See ToolOptions for a more detailed description.

Parameters:

Name	Type	Description	Default
app	FastAPI	The FastAPI context	required
path_prefix	str	The path prefix to use for this set of endpoints	required
worker_fn	Callable[[BaseModel, Optional[ExecutionContext], Optional[Response]], BaseModel]	description	required
opts	Optional[ToolOptions]	Additional behaviour settings. Defaults to ToolOptions().	ToolOptions()
context	Optional[ExecutionContext]	An optional context to be provided to every invocation of worker_fn. Defaults to None.	None

Source code in ivcap_lambda/builder.py

def add_tool_api_route(
    app: FastAPI,
    path_prefix: str,
    worker_fn: WorkerFn,
    *,
    opts: ToolOptions | None = ToolOptions(),
    context: ExecutionContext | None = None,
):
    """Add a few routes to `app` for use with an AI tool.

    The tool itself is implemented in `worker_fn` where the first
    argument is a pydantic model describing all the tool's "public" parameters.
    The function is also expected to return it's result as a single pydantic model.
    The tool function can have two optioanl paramters, one with the same type as
    `context` and the second one with `fastapi.Request`. The context paramter will
    be identical to the above `context`, while the `request` will be the incoming
    request.

    This function then sets up three endpoints:
    - POST {path_prefix}: To request the execution of the tool (the 'job')
    - GET {path_prefix}/{job_id}: To collect the result of a tool execution
    - GET {path_prefix}: To obtain a description of the tool suitable for most agent frameworks

    The POST request will only wait `opts.max_wait_time` for the tool to finish. If it
    hasn't finished by then, a `204 No Content` code will be returned with additional
    header fields `Location` and `Retry-later` to inform the caller where and approx.
    when the result can be collected later.

    The `opts` parameter allows for customization of the endpoints. See `ToolOptions`
    for a more detailed description.

    Args:
        app (FastAPI): The FastAPI context
        path_prefix (str): The path prefix to use for this set of endpoints
        worker_fn (Callable[[BaseModel, Optional[ExecutionContext], Optional[Response]], BaseModel]): _description_
        opts (Optional[ToolOptions], optional): Additional behaviour settings. Defaults to ToolOptions().
        context (Optional[ExecutionContext], optional): An optional context to be provided to every invocation of `worker_fn`. Defaults to None.
    """
    def_name, def_tag = get_title_from_path(path_prefix)
    if opts is None:
        opts = ToolOptions()
    if opts.tags is None:
        if def_tag == "":
            def_tag = "Tool"
        opts.tags = [def_tag]
    if opts.name is None:
        if def_name == "":
            def_name = "Execute the tool"
        opts.name = def_name

    output_model = get_function_return_type(worker_fn)
    executor = Executor[output_model](  # type: ignore[valid-type]
        worker_fn, opts=opts.executor_opts, context=context
    )

    tools.append(
        ToolDescription(
            name=worker_fn.__name__,
            path_prefix=path_prefix,
            worker_fn=worker_fn,
            input=get_input_type(worker_fn),
            executor=executor,
        )
    )

    add_ready_handler(opts.is_ready if opts else None)

    _add_do_job_route(app, path_prefix, worker_fn, executor, opts)
    _add_get_job_route(app, path_prefix, worker_fn, executor, opts)
    _add_get_tool_def_route(app, path_prefix, worker_fn, opts)

Copy Markdown