Skip to content

JobContext

JobContext is the gateway to the IVCAP platform from within your job worker function. It is passed automatically as the second argument to every worker function and provides:

  • Job identity — the unique ID and platform-injected authorisation token for the current job.
  • Progress & event reporting — a pre-configured EventReporter for emitting structured step events back to the platform.
  • Full IVCAP platform access — a lazy-initialized, pre-authenticated IVCAP client from the ivcap-client library for uploading/downloading artifacts, querying metadata, and calling other services.
from ivcap_service import JobContext

def process_job(req: MyRequest, ctxt: JobContext) -> MyResult:
    # Identity
    logger.info(f"Processing job {ctxt.job_id}")

    # Progress reporting
    with ctxt.report.step("analysis", message="Analysing input") as step:
        result = analyse(req.data)
        step.finished(message="Analysis complete")

    # Platform access
    artifact = ctxt.ivcap.get_artifact(req.input_urn)
    with artifact.as_local_file() as path:
        data = path.read_bytes()

    return MyResult(output=result)

Properties

job_id · str

The platform-assigned unique identifier for this job (e.g. urn:ivcap:job:...). Use this in log messages and when correlating events or results.

logger.info(f"Starting job {ctxt.job_id}")

report · EventReporter

A fully configured EventReporter instance for emitting structured progress events during execution. The reporter uses the platform sidecar when running on IVCAP, and falls back to structured logging locally.

Quick reference:

# Context-manager style (recommended)
with ctxt.report.step("load", message="Loading data") as step:
    data = load(req.source)
    step.info({"progress_percent": 50})   # (1)
    step.finished(message=f"Loaded {len(data)} records")

# Direct-call style (for async or conditional reporting, outside a `with` block)
ctxt.report.step_started("analysis", message="Starting analysis")
ctxt.report.step_finished("analysis", message="Done")
  1. step.info(...) is a method on the step object returned by ctxt.report.step(...) (inside the with block), not on the EventReporter itself. It accepts either a raw dict (wrapped into a GenericEvent) or any BaseEvent instance.

For full details on event types, the EventContext step object, and custom reporters see Events & Reporting API →

ivcap · IVCAP

A lazy-initialized, pre-authenticated client from the ivcap-client library. It is instantiated on first access — no credentials or configuration are required inside a platform job container.

ivcap = ctxt.ivcap   # property access, no parentheses

The client is the primary entry point for all platform interactions:

What you want to do Method / starting point
Download an input artifact ivcap.get_artifact(urn)
Upload a result artifact ivcap.upload_artifact(...)
Attach metadata to an entity ivcap.add_aspect(entity, aspect)
List or query existing aspects ivcap.list_aspects(entity=...)
Call another IVCAP service ivcap.get_service_by_name(...) → .request_job(...)
List available services ivcap.list_services(...)

Working with artifacts → Artifacts Guide

Calling other services → Service Composition Guide

job_authorization · str | None

The Bearer token injected by the platform for the current job. This is used internally by the SDK to authenticate outbound HTTP calls (via the requests and httpx monkey-patches in context.py) and by the EventReporter when posting events to the sidecar.

You rarely need to access this directly — the ivcap client and the HTTP instrumentation layer use it automatically.

Execution flow

sequenceDiagram
    participant P as IVCAP Platform
    participant S as Service Worker
    participant R as EventReporter
    participant C as IVCAP Client

    P->>S: invoke worker(req, ctxt)
    S->>R: ctxt.report.step("…")
    R->>P: POST /events/{job_id}
    S->>C: ctxt.ivcap.get_artifact(…)
    C->>P: GET /artifacts/{id}
    P-->>C: artifact stream
    S->>R: step.finished(…)
    R->>P: POST /events/{job_id}
    S->>P: result (via push_result)

API Reference

JobContext

Bases: BaseModel

Source code in ivcap_service/types.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class JobContext(BaseModel):
    job_id: str
    report: EventReporter
    job_authorization: str | None = None

    _ivcap: IVCAP | None = PrivateAttr(None)

    model_config = ConfigDict(arbitrary_types_allowed=True)

    @property
    def ivcap(self) -> IVCAP:
        if self._ivcap is None:
            self._ivcap = cast(IVCAP, IVCAP())
        return self._ivcap

    def llm_client(self, **kwargs: Any) -> "openai.OpenAI":
        """Return an OpenAI-compatible client, ready to use.

        This hides the details of how/whether to route completions through a
        LiteLLM proxy from the service code, and resolves the API key in the
        following order (first match wins):

        1. An explicit 'api_key' kwarg passed in here.
        2. 'LITELLM_PROXY' environment variable - if set, 'base_url' is
           pointed at that proxy's '/v1' endpoint and a placeholder
           'api_key' is used (the proxy itself handles
           authentication/authorization), so no real key is needed.
        3. 'OPENAI_API_KEY' environment variable - the usual variable read
           by a plain 'openai.OpenAI()' client ('OPENAI_BASE_URL' is used
           the same way for the endpoint).
        4. '~/.config/openai/api_key' - a plain text file containing just
           the key, as a local-development fallback for when you don't want
           to export environment variables.

        If none of the above yields a key (and 'LITELLM_PROXY' isn't set,
        since no key is needed in that case), a 'RuntimeError' is raised
        with guidance on how to configure one.

        Any 'kwargs' passed in are forwarded to 'openai.OpenAI(...)' and take
        precedence over everything above (e.g. pass 'base_url=...' to
        override the endpoint too).

        See also 'DEFAULT_LLM_MODEL' for the default model name to pass to
        e.g. 'client.chat.completions.create(model=..., ...)' when the
        caller/request doesn't specify one.

        Requires the optional 'openai' package to be installed.
        """
        try:
            import openai
        except ImportError as e:
            raise ImportError(
                "The 'openai' package is required to use JobContext.llm_client(). "
                "Install it with e.g. 'pip install openai'."
            ) from e

        base_url = os.getenv("LITELLM_PROXY")
        if base_url is not None:
            kwargs.setdefault("base_url", f"{base_url}/v1")
            kwargs.setdefault("api_key", "not-needed")
        elif "api_key" not in kwargs and not os.getenv("OPENAI_API_KEY"):
            api_key = _read_openai_api_key_file()
            if api_key is None:
                raise RuntimeError(
                    "No OpenAI API key found. Set the 'OPENAI_API_KEY' environment "
                    f"variable, write the key to '{_OPENAI_API_KEY_FILE}', or pass "
                    "'api_key=...' to 'llm_client()' directly."
                )
            kwargs["api_key"] = api_key
        return openai.OpenAI(**kwargs)

llm_client(**kwargs)

Return an OpenAI-compatible client, ready to use.

This hides the details of how/whether to route completions through a LiteLLM proxy from the service code, and resolves the API key in the following order (first match wins):

  1. An explicit 'api_key' kwarg passed in here.
  2. 'LITELLM_PROXY' environment variable - if set, 'base_url' is pointed at that proxy's '/v1' endpoint and a placeholder 'api_key' is used (the proxy itself handles authentication/authorization), so no real key is needed.
  3. 'OPENAI_API_KEY' environment variable - the usual variable read by a plain 'openai.OpenAI()' client ('OPENAI_BASE_URL' is used the same way for the endpoint).
  4. '~/.config/openai/api_key' - a plain text file containing just the key, as a local-development fallback for when you don't want to export environment variables.

If none of the above yields a key (and 'LITELLM_PROXY' isn't set, since no key is needed in that case), a 'RuntimeError' is raised with guidance on how to configure one.

Any 'kwargs' passed in are forwarded to 'openai.OpenAI(...)' and take precedence over everything above (e.g. pass 'base_url=...' to override the endpoint too).

See also 'DEFAULT_LLM_MODEL' for the default model name to pass to e.g. 'client.chat.completions.create(model=..., ...)' when the caller/request doesn't specify one.

Requires the optional 'openai' package to be installed.

Source code in ivcap_service/types.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def llm_client(self, **kwargs: Any) -> "openai.OpenAI":
    """Return an OpenAI-compatible client, ready to use.

    This hides the details of how/whether to route completions through a
    LiteLLM proxy from the service code, and resolves the API key in the
    following order (first match wins):

    1. An explicit 'api_key' kwarg passed in here.
    2. 'LITELLM_PROXY' environment variable - if set, 'base_url' is
       pointed at that proxy's '/v1' endpoint and a placeholder
       'api_key' is used (the proxy itself handles
       authentication/authorization), so no real key is needed.
    3. 'OPENAI_API_KEY' environment variable - the usual variable read
       by a plain 'openai.OpenAI()' client ('OPENAI_BASE_URL' is used
       the same way for the endpoint).
    4. '~/.config/openai/api_key' - a plain text file containing just
       the key, as a local-development fallback for when you don't want
       to export environment variables.

    If none of the above yields a key (and 'LITELLM_PROXY' isn't set,
    since no key is needed in that case), a 'RuntimeError' is raised
    with guidance on how to configure one.

    Any 'kwargs' passed in are forwarded to 'openai.OpenAI(...)' and take
    precedence over everything above (e.g. pass 'base_url=...' to
    override the endpoint too).

    See also 'DEFAULT_LLM_MODEL' for the default model name to pass to
    e.g. 'client.chat.completions.create(model=..., ...)' when the
    caller/request doesn't specify one.

    Requires the optional 'openai' package to be installed.
    """
    try:
        import openai
    except ImportError as e:
        raise ImportError(
            "The 'openai' package is required to use JobContext.llm_client(). "
            "Install it with e.g. 'pip install openai'."
        ) from e

    base_url = os.getenv("LITELLM_PROXY")
    if base_url is not None:
        kwargs.setdefault("base_url", f"{base_url}/v1")
        kwargs.setdefault("api_key", "not-needed")
    elif "api_key" not in kwargs and not os.getenv("OPENAI_API_KEY"):
        api_key = _read_openai_api_key_file()
        if api_key is None:
            raise RuntimeError(
                "No OpenAI API key found. Set the 'OPENAI_API_KEY' environment "
                f"variable, write the key to '{_OPENAI_API_KEY_FILE}', or pass "
                "'api_key=...' to 'llm_client()' directly."
            )
        kwargs["api_key"] = api_key
    return openai.OpenAI(**kwargs)