Executor & Options

This page documents ToolOptions, ExecutorOpts, and ExecutionContext — the classes used to configure how tool functions are executed.

ToolOptions

ToolOptions

Bases: BaseModel

Source code in ivcap_lambda/builder.py

class ToolOptions(BaseModel):
    name: str | None = Field(None, description="Name to be used for this tool")
    tags: list[str] | None = Field(
        None, description="OpenAPI tag for this set of functions"
    )
    max_wait_time: float | None = Field(
        5.0,
        description="max. time in seconds to wait for result and before returning RetryLater",
    )
    refresh_interval: int | None = Field(
        3,
        description="Time in seconds to wait before chacking again for a job result (used in RetryLater)",
    )
    executor_opts: ExecutorOpts | None = Field(
        None, description="Options for the executor"
    )
    post_route_opts: dict[str, Any] | None = Field(
        {}, description="Addtitional options given the POST route constructor"
    )
    service_id: str | None = Field(
        None, description="overriding the default service id"
    )
    is_ready: Callable[[], bool] | None = Field(
        None, description="Function to check if the tool is ready"
    )

---

ExecutorOpts

ExecutorOpts

Bases: BaseModel

Source code in ivcap_lambda/executor.py

class ExecutorOpts(BaseModel):
    job_cache_size: int | None = Field(10000, description="size of job cache")
    job_cache_ttl: int | None = Field(
        3600, description="TTL of job entries in the job cache"
    )
    max_workers: int | None = Field(
        None,
        description="size of thread pool to use. If None, a new thread pool will be created for each execution",
    )

---

ExecutionContext

ExecutionContext

Source code in ivcap_lambda/executor.py

class ExecutionContext:
    pass

---

Executor (internal)

The Executor class manages the thread pool and job result cache. It is created automatically by add_tool_api_route based on ToolOptions.executor_opts. You do not normally need to interact with it directly.

Executor

Bases: Generic[T]

A generic class that executes a function in a thread pool and returns the result via an asyncio Queue. The generic type T represents the return type of the function.

Source code in ivcap_lambda/executor.py

class Executor(Generic[T]):
    """
    A generic class that executes a function in a thread pool and returns the result via an asyncio Queue.
    The generic type T represents the return type of the function.
    """

    # _job_ctxt = JobContext()
    _active_jobs: set[str] = (
        set()
    )  # keep track of active jobs to block shutdown until they are done

    @classmethod
    def active_jobs(cls) -> list[str]:
        """Returns a list of IDs of the currently active jobs"""
        return list(cls._active_jobs)

    @classmethod
    def wait_for_exit_ready(cls):
        """The server is calling this method when a shutdown request arrived. It will
        proceed with the shutdown when this method returns.

        We may implement functionality to only return when all active jobs have finsihed, as well as not
        accepting any new incoming requests.
        """
        while len(cls._active_jobs) > 0:
            logger.info(
                f"blocking shutdown as {len(cls._active_jobs)} job(s) are still running"
            )
            sleep(5)
        return

    def __init__(
        self,
        func: Callable[..., T],
        *,
        opts: ExecutorOpts | None,
        context: ExecutionContext | None = None,
    ):
        """
        Initialize the Executor with a function and an optional thread pool.

        Args:
            func: The function to execute, returning type T
            opts:
             - job_cache_size: Optional size of job cache. Defaults to 1000
             - job_cache_ttl: Optional TTL of job entries in the job cache. Defaults to 600 sec
             - max_workers: Optional size of thread pool to use. If None, a new thread pool will be created for each execution.
        """
        self.func = func
        if opts is None:
            opts = ExecutorOpts()
        self.job_cache: TTLCache = TTLCache(
            maxsize=opts.job_cache_size or 10000, ttl=opts.job_cache_ttl or 3600
        )
        self.thread_pool = None
        if opts.max_workers:
            self.thread_pool = concurrent.futures.ThreadPoolExecutor(
                max_workers=opts.max_workers
            )

        self.context = context
        self.context_param = None
        self.request_param = None
        self.job_ctxt_param = None
        _, extras = get_input_type(func)
        for k, v in extras.items():
            if isinstance(context, v):
                self.context_param = k
            elif v == Request:
                self.request_param = k
            elif v == JobContext:
                self.job_ctxt_param = k
            else:
                raise Exception(f"unexpected function parameter '{k}'")

    async def execute(
        self,
        param: Any,
        job_id: str,
        req: Request,
        report_result: bool = True,
        reporter: EventReporter | None = None,
    ) -> asyncio.Queue[T | ExecutionError]:
        """
        Execute the function with the given parameter in a thread and return a queue with the result.

        Args:
            param: Any The parameter to pass to the function
            job_id: str ID of this job
            req: Request FastAPI's request object (or any duck-typed object
                exposing a `.headers.get(...)` mapping, e.g. an MCP request shim)
            report_result: whether to push the result back to the IVCAP sidecar
            reporter: optional `EventReporter` instance to use for this invocation
                instead of the one produced by the globally configured event-reporter
                factory (e.g. `SidecarReporter`). Used to bridge progress reporting to
                non-REST transports such as MCP.

        Returns:
            An asyncio Queue that will contain either the result of type T or an ExecutionError
        """
        result_queue: asyncio.Queue[T | ExecutionError] = asyncio.Queue()
        event_loop = asyncio.get_running_loop()
        self.job_cache[job_id] = None

        def _process_result(result):
            """Verify the result, add it to the queue, and report it to IVCAP."""
            try:
                result = verify_result(result, job_id, logger)
            except Exception as e:
                result = ExecutionError(
                    error=str(e),
                    type=type(e).__name__,
                    traceback=traceback.format_exc(),
                )
                logger.warning(f"job {job_id} failed - {result.error}")
            finally:
                self.job_cache[job_id] = result
                logger.info(f"job {job_id} finished, sending result message")
                asyncio.run_coroutine_threadsafe(
                    result_queue.put(result),
                    event_loop,
                )
                if report_result:
                    push_result(result, job_id)
                self.__class__._active_jobs.discard(job_id)

        def _run(param: Any, ctxt: Context):
            context.attach(ctxt)  # OTEL
            authorization = req.headers.get("authorization")
            jctxt = JobContext(
                job_id=job_id,
                job_authorization=authorization,
                report=reporter
                if reporter is not None
                else create_event_reporter(
                    job_id=job_id, job_authorization=authorization
                ),
            )
            job_context.set(jctxt)
            kwargs: dict[str, Any] = {}
            if self.context_param is not None:
                kwargs[self.context_param] = self.context
            if self.request_param is not None:
                kwargs[self.request_param] = req
            if self.job_ctxt_param is not None:
                kwargs[self.job_ctxt_param] = jctxt  # self._job_ctxt

            fname = self.func.__name__
            with tracer.start_as_current_span(f"RUN {fname}") as span:
                span.set_attribute("job.id", job_id)
                span.set_attribute("job.name", fname)
                loop = None
                try:
                    self.__class__._active_jobs.add(job_id)
                    if asyncio.iscoroutinefunction(self.func):
                        loop = asyncio.new_event_loop()
                        asyncio.set_event_loop(loop)
                        try:
                            res = loop.run_until_complete(self.func(param, **kwargs))
                        except (asyncio.CancelledError, GeneratorExit):
                            # Propagate cancellation and generator shutdown properly
                            raise
                        finally:
                            # Gracefully shutdown remaining tasks and async generators to avoid
                            # 'Task exception was never retrieved' and similar warnings
                            try:
                                pending = [
                                    t for t in asyncio.all_tasks(loop) if not t.done()
                                ]
                                for t in pending:
                                    t.cancel()
                                if pending:
                                    with contextlib.suppress(Exception):
                                        loop.run_until_complete(
                                            asyncio.gather(
                                                *pending, return_exceptions=True
                                            )
                                        )
                                with contextlib.suppress(Exception):
                                    loop.run_until_complete(loop.shutdown_asyncgens())
                            finally:
                                loop.close()
                    else:
                        res = self.func(param, **kwargs)
                except (asyncio.CancelledError, GeneratorExit):
                    # Allow cooperative shutdown/cancellation to propagate cleanly
                    span.record_exception(Exception("cancelled"))
                    raise
                except Exception as ex:
                    span.record_exception(ex)
                    logger.error(
                        f"while executing {job_id} - {type(ex).__name__}: {ex}"
                    )
                    res = ExecutionError(
                        error=str(ex),
                        type=type(ex).__name__,
                        traceback=traceback.format_exc(),
                    )
                finally:
                    self.__class__._active_jobs.discard(job_id)

                try:
                    _process_result(res)
                except Exception as ex:
                    logger.error(f"while delivering result fo {job_id} - {ex}")

                job_context.set(None)

        # Use the provided thread pool or create a new one
        use_pool = self.thread_pool or concurrent.futures.ThreadPoolExecutor(
            max_workers=1
        )
        # Submit the function to the thread pool
        future = use_pool.submit(_run, param, context.get_current())

        # If we created a new pool, we should clean it up when done
        if self.thread_pool is None:
            future.add_done_callback(lambda _: use_pool.shutdown(wait=False))
        return result_queue

    def lookup_job(self, job_id: str) -> T | ExecutionError | None:
        """Return the result of a job

        Args:
            job_id (str): The id of the job requested

        Returns:
            Union[T, ExecutionError, None]: Returns the result fo a job, 'None' is still in progress

        Raises:
            KeyError: Unknown job - may have already expired
        """
        return self.job_cache[job_id]

active_jobs() classmethod

Returns a list of IDs of the currently active jobs

Source code in ivcap_lambda/executor.py

@classmethod
def active_jobs(cls) -> list[str]:
    """Returns a list of IDs of the currently active jobs"""
    return list(cls._active_jobs)

wait_for_exit_ready() classmethod

The server is calling this method when a shutdown request arrived. It will proceed with the shutdown when this method returns.

We may implement functionality to only return when all active jobs have finsihed, as well as not accepting any new incoming requests.

Source code in ivcap_lambda/executor.py

@classmethod
def wait_for_exit_ready(cls):
    """The server is calling this method when a shutdown request arrived. It will
    proceed with the shutdown when this method returns.

    We may implement functionality to only return when all active jobs have finsihed, as well as not
    accepting any new incoming requests.
    """
    while len(cls._active_jobs) > 0:
        logger.info(
            f"blocking shutdown as {len(cls._active_jobs)} job(s) are still running"
        )
        sleep(5)
    return

execute(param, job_id, req, report_result=True, reporter=None) async

Execute the function with the given parameter in a thread and return a queue with the result.

Parameters:

Name	Type	Description	Default
param	Any	Any The parameter to pass to the function	required
job_id	str	str ID of this job	required
req	Request	Request FastAPI's request object (or any duck-typed object exposing a .headers.get(...) mapping, e.g. an MCP request shim)	required
report_result	bool	whether to push the result back to the IVCAP sidecar	True
reporter	EventReporter | None	optional EventReporter instance to use for this invocation instead of the one produced by the globally configured event-reporter factory (e.g. SidecarReporter). Used to bridge progress reporting to non-REST transports such as MCP.	None

Returns:

Type	Description
Queue[T | ExecutionError]	An asyncio Queue that will contain either the result of type T or an ExecutionError

Source code in ivcap_lambda/executor.py

async def execute(
    self,
    param: Any,
    job_id: str,
    req: Request,
    report_result: bool = True,
    reporter: EventReporter | None = None,
) -> asyncio.Queue[T | ExecutionError]:
    """
    Execute the function with the given parameter in a thread and return a queue with the result.

    Args:
        param: Any The parameter to pass to the function
        job_id: str ID of this job
        req: Request FastAPI's request object (or any duck-typed object
            exposing a `.headers.get(...)` mapping, e.g. an MCP request shim)
        report_result: whether to push the result back to the IVCAP sidecar
        reporter: optional `EventReporter` instance to use for this invocation
            instead of the one produced by the globally configured event-reporter
            factory (e.g. `SidecarReporter`). Used to bridge progress reporting to
            non-REST transports such as MCP.

    Returns:
        An asyncio Queue that will contain either the result of type T or an ExecutionError
    """
    result_queue: asyncio.Queue[T | ExecutionError] = asyncio.Queue()
    event_loop = asyncio.get_running_loop()
    self.job_cache[job_id] = None

    def _process_result(result):
        """Verify the result, add it to the queue, and report it to IVCAP."""
        try:
            result = verify_result(result, job_id, logger)
        except Exception as e:
            result = ExecutionError(
                error=str(e),
                type=type(e).__name__,
                traceback=traceback.format_exc(),
            )
            logger.warning(f"job {job_id} failed - {result.error}")
        finally:
            self.job_cache[job_id] = result
            logger.info(f"job {job_id} finished, sending result message")
            asyncio.run_coroutine_threadsafe(
                result_queue.put(result),
                event_loop,
            )
            if report_result:
                push_result(result, job_id)
            self.__class__._active_jobs.discard(job_id)

    def _run(param: Any, ctxt: Context):
        context.attach(ctxt)  # OTEL
        authorization = req.headers.get("authorization")
        jctxt = JobContext(
            job_id=job_id,
            job_authorization=authorization,
            report=reporter
            if reporter is not None
            else create_event_reporter(
                job_id=job_id, job_authorization=authorization
            ),
        )
        job_context.set(jctxt)
        kwargs: dict[str, Any] = {}
        if self.context_param is not None:
            kwargs[self.context_param] = self.context
        if self.request_param is not None:
            kwargs[self.request_param] = req
        if self.job_ctxt_param is not None:
            kwargs[self.job_ctxt_param] = jctxt  # self._job_ctxt

        fname = self.func.__name__
        with tracer.start_as_current_span(f"RUN {fname}") as span:
            span.set_attribute("job.id", job_id)
            span.set_attribute("job.name", fname)
            loop = None
            try:
                self.__class__._active_jobs.add(job_id)
                if asyncio.iscoroutinefunction(self.func):
                    loop = asyncio.new_event_loop()
                    asyncio.set_event_loop(loop)
                    try:
                        res = loop.run_until_complete(self.func(param, **kwargs))
                    except (asyncio.CancelledError, GeneratorExit):
                        # Propagate cancellation and generator shutdown properly
                        raise
                    finally:
                        # Gracefully shutdown remaining tasks and async generators to avoid
                        # 'Task exception was never retrieved' and similar warnings
                        try:
                            pending = [
                                t for t in asyncio.all_tasks(loop) if not t.done()
                            ]
                            for t in pending:
                                t.cancel()
                            if pending:
                                with contextlib.suppress(Exception):
                                    loop.run_until_complete(
                                        asyncio.gather(
                                            *pending, return_exceptions=True
                                        )
                                    )
                            with contextlib.suppress(Exception):
                                loop.run_until_complete(loop.shutdown_asyncgens())
                        finally:
                            loop.close()
                else:
                    res = self.func(param, **kwargs)
            except (asyncio.CancelledError, GeneratorExit):
                # Allow cooperative shutdown/cancellation to propagate cleanly
                span.record_exception(Exception("cancelled"))
                raise
            except Exception as ex:
                span.record_exception(ex)
                logger.error(
                    f"while executing {job_id} - {type(ex).__name__}: {ex}"
                )
                res = ExecutionError(
                    error=str(ex),
                    type=type(ex).__name__,
                    traceback=traceback.format_exc(),
                )
            finally:
                self.__class__._active_jobs.discard(job_id)

            try:
                _process_result(res)
            except Exception as ex:
                logger.error(f"while delivering result fo {job_id} - {ex}")

            job_context.set(None)

    # Use the provided thread pool or create a new one
    use_pool = self.thread_pool or concurrent.futures.ThreadPoolExecutor(
        max_workers=1
    )
    # Submit the function to the thread pool
    future = use_pool.submit(_run, param, context.get_current())

    # If we created a new pool, we should clean it up when done
    if self.thread_pool is None:
        future.add_done_callback(lambda _: use_pool.shutdown(wait=False))
    return result_queue

lookup_job(job_id)

Return the result of a job

Parameters:

Name	Type	Description	Default
job_id	str	The id of the job requested	required

Returns:

Type	Description
T | ExecutionError | None	Union[T, ExecutionError, None]: Returns the result fo a job, 'None' is still in progress

Raises:

Type	Description
KeyError	Unknown job - may have already expired

Source code in ivcap_lambda/executor.py

def lookup_job(self, job_id: str) -> T | ExecutionError | None:
    """Return the result of a job

    Args:
        job_id (str): The id of the job requested

    Returns:
        Union[T, ExecutionError, None]: Returns the result fo a job, 'None' is still in progress

    Raises:
        KeyError: Unknown job - may have already expired
    """
    return self.job_cache[job_id]

Copy Markdown