Skip to main content
The Orca runtime injects an AgentContext object into every invocation so your agent code can read message metadata, publish responses, and manage runtime state. This reference summarizes the key APIs that become available after importing AgentContext in your project: 
from orca.functions.agent_context import AgentContext

​
Access the current context

  • AgentContext.current() returns the context bound to the active asynchronous task or thread. It raises RuntimeError if your code runs outside an initialized request scope.
  • AgentContext.enter(ctx) explicitly associates a context with the current task. Use it when you spawn your own threads or background jobs so downstream code can call AgentContext.current() safely.
  • The context implements synchronous and asynchronous context managers. Wrap long-running sections in with AgentContext.current() as ctx: (or async with inside asynchronous code) to ensure the previous context is restored automatically.

​
Message metadata and acknowledgement

The runtime records properties of the active message before your agent executes. Use these helpers to inspect request details:
  • get_message_id() exposes the origin of the current message.
  • get_message_key(), get_partition_key(), get_message_eventtime(), and get_message_properties() surface custom metadata that producers may include.
  • get_current_message_topic_name() returns the input topic name for the current invocation.
  • get_function_name(), get_function_tenant(), get_function_namespace(), get_function_id(), get_function_version(), and get_instance_id() identify the deployed agent and running instance.
To emit outputs or acknowledge work:
  • publish(topic_name, message, serde_class_name="serde.IdentitySerDe", properties=None, compression_type=None, callback=None, message_conf=None) sends asynchronous responses. The context reuses producers under the hood and accepts extra message configuration through the optional message_conf dictionary.
  • ack(msgid=None, topic=None) acknowledges the active message. Provide a msgid or topic to override the defaults when you manage acknowledgements manually.

​
Configuration, secrets, and logging

  • get_user_config_value(key) and get_user_config_map() expose agent-level configuration supplied at deployment time.
  • get_secret(secret_key) retrieves secrets from the configured provider. Expect a None return value when a key is missing and handle defaults accordingly.
  • get_logger() returns the runtime logger so you can emit structured logs that align with platform observability settings.

​
Metrics, state, and counters

AgentContext provides utilities for custom telemetry and lightweight state management:
  • record_metric(metric_name, metric_value) records Prometheus summary metrics with the current instance labels.
  • incr_counter(key, amount), get_counter(key), and del_counter(key) maintain numeric counters.
  • put_state(key, value) and get_state(key) persist arbitrary state using the runtime-backed store.

​
Topic configuration

  • get_input_topics() lists all input topics connected to the agent.
  • get_output_topic() and get_output_serde_class_name() show where default outputs go and which serialization class the runtime expects.

​
Session management

  • get_session_mode() returns the configured session strategy:
    • SessionMode.SHARED reuses one conversation for every invocation.
    • SessionMode.SESSION_PER_MESSAGE isolates each request.
    • SessionMode.SESSION_PER_USER partitions sessions by user identity when the incoming messages include user identifiers.
  • For tool discovery and Model Context Protocol integrations, see Managed agent tools.

​
Example usage

The snippet below shows how an agent can inspect message metadata, record metrics, and publish follow-up events while acknowledging work through the context: 
from orca.functions.agent_context import AgentContext

async def handle_request(payload: dict) -> str:
    ctx = AgentContext.current()
    logger = ctx.get_logger()

    message_id = ctx.get_message_id()
    logger.info("processing message %s", message_id)

    ctx.record_metric("agent_requests_total", 1)
    routing_mode = ctx.get_user_config_value("routing-mode") or "direct"

    if routing_mode == "fanout":
        ctx.publish(ctx.get_output_topic(), {"status": "received", "payload": payload})

    ctx.ack()
    return f"handled message {message_id}"

​
Best practices

  • Always call AgentContext.current() inside request handlers so you stay within the active invocation scope.
  • When launching background tasks, copy the active context with AgentContext.enter(context_instance) before executing work.
  • Handle None returns from get_user_config_value and get_secret gracefully to keep agents resilient to missing settings.
  • Use the metrics helpers sparingly to avoid high-cardinality metric names, and reset counters when they no longer apply.
  • Publish follow-up events and acknowledge messages through the context so the runtime can reuse producers and maintain consistent delivery guarantees.
With these APIs in mind, you can build agents that combine messaging, durable state, and session-aware behavior across the Orca Engine platform.