Building Execution Engines

Execution engines let you run Ark agents with custom frameworks (LangChain, CrewAI, etc.) while integrating with Ark’s resource management. They receive agent context via the A2A v0.3.0 query extension  and return results over the A2A protocol.

For hosting raw A2A agents without Ark integration, see Building A2A Servers.

Quickstart

Install the Ark SDK:

pip install ark-sdk

Create a minimal executor:

from ark_sdk.executor import BaseExecutor, ExecutionEngineRequest, Message
from ark_sdk.executor_app import ExecutorApp
 
 
class MyExecutor(BaseExecutor):
    async def execute_agent(self, request: ExecutionEngineRequest) -> list[Message]:
        # request.agent has the agent config (name, prompt, model, parameters)
        # request.userInput has the user message
        # request.mcpServers has MCP server connection info (url, transport, headers, tool allowlist)
        # request.conversationId has the conversation threading ID (from A2A contextId)
        prompt = self._resolve_prompt(request.agent)
        return [Message(role="assistant", content=f"Echo: {request.userInput.content}")]
 
 
app = ExecutorApp(
    executor=MyExecutor("my-engine"),
    engine_name="my-engine",
    description="A simple execution engine",
)
 
if __name__ == "__main__":
    app.run()

Run it:

python main.py

The engine serves an agent card at /.well-known/agent-card.json that declares support for the Ark query extension.

How It Works

When Ark dispatches a query to your engine:

1. The A2A message carries only a QueryRef (name + namespace) via the query extension — no agent config, credentials, or tool definitions cross the wire
2. The SDK extracts the QueryRef and the A2A contextId (conversation threading)
3. Within your executor pod, the SDK fetches the Query CRD and resolves agent config, model (including API keys), and MCP servers from the cluster
4. Your execute_agent() receives a fully populated ExecutionEngineRequest with conversationId

Secrets never leave the cluster over A2A — they are resolved locally within your executor pod. You never interact with the extension or Kubernetes APIs directly.

Resource resolution runs as the Query’s spec.serviceAccount when it declares one, and as your executor pod’s service account otherwise. Queries that declare one need your executor’s service account to hold impersonate on serviceaccounts; without it the query fails rather than silently resolving with the pod identity.

Scope that grant with resourceNames to the accounts queries may name. Granting it unscoped lets anything holding your executor’s token act as any service account in the namespace.

Extension Spec

The query extension contract is defined at ark/api/extensions/query/v1/. The schema is minimal: name and namespace identifying the Query resource, plus an optional target (see below).

Deploying to Kubernetes

Create an ExecutionEngine resource pointing to your engine:

apiVersion: ark.mckinsey.com/v1prealpha1
kind: ExecutionEngine
metadata:
  name: my-engine
spec:
  address:
    value: "http://my-engine-service:8000"

Then reference it from an Agent:

apiVersion: ark.mckinsey.com/v1alpha1
kind: Agent
metadata:
  name: my-agent
spec:
  prompt: "You are a helpful assistant."
  executionEngine:
    name: my-engine
  modelRef:
    name: my-model

Queries targeting my-agent will be routed to your engine via A2A.

Streaming Support

If your engine writes streaming chunks to the broker (via the memory event stream), opt in by adding the streaming-supported annotation to your ExecutionEngine resource:

apiVersion: ark.mckinsey.com/v1prealpha1
kind: ExecutionEngine
metadata:
  name: my-engine
  annotations:
    ark.mckinsey.com/streaming-supported: "true"
spec:
  address:
    value: "http://my-engine-service:8000"

Without this annotation, streaming requests are handled transparently — clients still receive a valid SSE response, just without incremental chunks.

Streaming text and tool calls

BaseExecutor provides two methods for pushing activity to the stream as it happens:

async def execute_agent(self, request):
    await self.stream_chunk("Looking that up...")
 
    await self.stream_tool_call(
        name="Bash",
        arguments={"command": "ls /tmp"},   # dict or pre-serialized JSON string
        tool_call_id="toolu_1",             # optional, from your framework
    )

stream_chunk() emits a text delta. stream_tool_call() emits an OpenAI-format delta.tool_calls chunk, the same shape the built-in completions engine produces, so the dashboard and CLI render engine tool activity without any client changes. Call it as each tool is invoked — an agent running a long tool loop otherwise appears idle until it produces text.

Both are no-ops when the broker is unavailable or the call is a team-member sub-request, so they are always safe to call.

Streaming state is scoped to the request, so concurrent queries handled by the same executor never share it. Child tasks inherit that state, so calling either method from asyncio.gather() or asyncio.create_task() works — including the streamed-text flag, which is visible to the caller afterwards.

Both methods are coroutines and must be awaited on the event loop. Do not call them from a worker thread: asyncio.to_thread() and a raw threading.Thread each produce an un-awaited coroutine, so nothing is sent and nothing is raised. If your engine does work in a thread, return the result and stream it from the async side.

Tool results are not streamed. They can be large and may contain secrets, and the built-in engine does not stream them either. Return them in your execute_agent() messages instead, where they reach the UI with the completed query.

Team Membership

An engine-backed Agent can be a member of a Team. Teams are orchestrated by the completions executor, which calls your engine once per member turn.

A member call differs from a top-level dispatch in three ways:

• The extension carries an explicit target. The Query targets the team, not the member, so the caller sends target: {type: agent, name: <member>} alongside the QueryRef. It overrides query.spec.target. The SDK resolves it for you, and execute_agent() receives the member’s ExecutionEngineRequest as usual.
• Your engine must not write to the query’s stream, memory or status. The calling completions engine owns all three for the whole team run; a second writer double-writes memory and churns status.phase. The SDK enforces this by disabling the broker client and the status updater whenever target is present — but if your engine writes query status independently of the SDK, gate it on target being absent.
• streaming-supported is ignored. Member calls use blocking message/send, because broker chunks are keyed by query name and would interleave across members.

The preceding team transcript is folded into userInput as text, so a sequential member sees what the members before it said. There is no separate history field on ExecutionEngineRequest.

Resolving target requires an ark-sdk release that supports it — any version from the release that introduced team members onwards. An older engine ignores the field, sees spec.target.type: team, and rejects the call, naming the ark-sdk version it is running. Top-level dispatches are unaffected, since they never carry a target.

Per-tenant built-in completions

The same model applies to the built-in completions engine. By default it runs once, centrally, in ark-system, and every built-in Agent (one with no executionEngine field) is routed there. For tenant isolation — independent scaling, NetworkPolicy/PodSecurity, resource governance, and especially per-tenant Azure Workload Identity — deploy the completions chart into the tenant namespace as an opt-in:

helm install ark-completions \
  oci://ghcr.io/mckinsey/agents-at-scale-ark/charts/ark-completions \
  -n tenant-a \
  --set serviceAccount.create=true \
  --set rbac.create=true \
  --set executionEngine.create=true \
  --set serviceAccount.annotations."azure\.workload\.identity/client-id"=<client-id> \
  --set podLabels."azure\.workload\.identity/use"=true

This deploys a completions pod under a tenant-local ServiceAccount (with the tenant’s workload identity), grants it namespace-scoped RBAC, and registers an ExecutionEngine named ark-completions in the tenant namespace. The controller’s dispatch then prefers that engine for the tenant’s built-in Agents, falling back to the central ark-system engine for any namespace without one. No Agent changes are required — the routing is by namespace.

This is fully backward-compatible: the central ark-completions keeps working as the default, and existing installs are unaffected.

Conversation Threading

When a query includes spec.conversationId, the controller sends it to your engine as the A2A contextId. The SDK extracts it and populates request.conversationId.

Use this for stateful engines that need multi-turn support:

class StatefulExecutor(BaseExecutor):
    async def execute_agent(self, request: ExecutionEngineRequest) -> list[Message]:
        session = self.get_or_create_session(request.conversationId)
        response = session.process(request.userInput.content)
        return [Message(role="assistant", content=response)]

If conversationId is empty, the query is a single-turn interaction. The conversation ID returned via A2A response is stored in status.conversationId for clients to use in follow-up queries.

Key Types

Import from ark_sdk.executor:

Import from ark_sdk.executor_app:

MCP Tools

When an agent has MCP-type tools, the SDK resolves the MCPServer CRDs and provides connection info via request.mcpServers. Each entry contains everything needed to establish an MCP client connection:

for server in request.mcpServers:
    # server.name      — MCPServer resource name
    # server.url       — resolved server address
    # server.transport — "http" or "sse"
    # server.timeout   — e.g. "30s"
    # server.headers   — resolved headers (secrets already dereferenced)
    # server.tools     — allowlist of tool names this agent can use

Your executor connects to each MCP server using its own MCP client (e.g., via LangChain, CrewAI, or the MCP SDK directly). The tools list is an allowlist — only register and use tools that appear in this list, even if the server exposes additional tools.

Non-MCP tool types (http, agent, team, builtin) are not included in mcpServers. MCP is the only tool type with a dialable endpoint, so there is nothing for an engine to invoke.

Attaching one anyway is accepted, but warns. Applying the agent prints an admission warning naming the engine and the dropped tools:

Warning: agent 'toolagent': execution engine 'my-engine' receives only mcp tools;
these tools will not be available to the agent: get-coordinates (http)

The dashboard shows the same warning in the agent’s chat panel, and the SDK logs it when it builds the request. Without this the tool is dropped silently and the agent answers that it does not exist.