Python SDK
Certified OpenAI Agents SDK and Claude Agent SDK recipes.
Stacklink supplies executable, governed tools. Your selected agent harness owns the model call, tool loop, and final answer.
uv add stacklink-sdkCreate a project API key in Stacklink Overview, keep it in STACKLINK_API_KEY, and use the complete recipe for your harness below. The API key is for code using the SDK; MCP app connections use OAuth instead.
OpenAI Agents SDK
Install the Stacklink SDK with the selected harness:
uv add "stacklink-sdk[openai-agents]"Required environment: STACKLINK_API_KEY, OPENAI_API_KEY, OPENAI_MODEL.
STACKLINK_BASE_URL is optional and defaults to https://api.stacklink.in.
import asyncio
import os
import json
import uuid
from agents import Agent, Runner
from stacklink import Stacklink, create_stacklink_openai_agents_tools
async def main() -> None:
stacklink = Stacklink(
api_key=require_environment("STACKLINK_API_KEY"),
base_url=os.getenv("STACKLINK_BASE_URL", "https://api.stacklink.in"),
)
session = stacklink.sessions.create_bound({
"externalTenantId": "acme",
"externalUserId": "user_123",
"modelSurface": {"mode": "router"},
})
try:
prompt = "Find the best Gmail tool for reading unread messages and explain the next required action."
recalled = await asyncio.to_thread(session.memory.context, {"query": prompt})
memory_context = recalled.get("promptText", "")
agent = Agent(
name="Stacklink assistant",
model=require_environment("OPENAI_MODEL"),
instructions=(
"Use Stacklink search, exact contracts, connection management, "
"and governed execution to complete the request. Keep the same retry key for the same action. If an action is pending, inspect that operation; never submit it again with a new key."
+ "\n\n" + memory_context
),
tools=create_stacklink_openai_agents_tools(session),
)
result = await Runner.run(
agent,
prompt,
)
if not result.interruptions:
await asyncio.to_thread(session.memory.record_conversation_turn, {
"applicationKey": "your-app",
"turnId": str(uuid.uuid4()),
"content": json.dumps({
"schemaVersion": "stacklink.conversation.turn.v1",
"user": {"role": "user", "content": prompt},
"assistant": {"role": "assistant", "content": str(result.final_output)},
}, ensure_ascii=False),
})
print(result.final_output)
finally:
# Keep the application session while an accepted action is still pending.
cursor = None
while True:
page = stacklink.executions.list(session.id, cursor=cursor, limit=100)
if page["liveCount"]:
print("Keep this session for the pending operation:", session.id)
break
cursor = page["nextCursor"]
if not cursor:
session.close()
break
def require_environment(name: str) -> str:
value = os.getenv(name)
if not value:
raise RuntimeError(f"{name} is required.")
return value
if __name__ == "__main__":
asyncio.run(main())Claude Agent SDK
Install the Stacklink SDK with the selected harness:
uv add "stacklink-sdk[claude-agent]"Required environment: STACKLINK_API_KEY, ANTHROPIC_API_KEY, CLAUDE_MODEL.
STACKLINK_BASE_URL is optional and defaults to https://api.stacklink.in.
import asyncio
import os
import json
import uuid
from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query
from stacklink import Stacklink, create_stacklink_claude_agent_tool_config
async def main() -> None:
stacklink = Stacklink(
api_key=require_environment("STACKLINK_API_KEY"),
base_url=os.getenv("STACKLINK_BASE_URL", "https://api.stacklink.in"),
)
session = stacklink.sessions.create_bound({
"externalTenantId": "acme",
"externalUserId": "user_123",
"modelSurface": {"mode": "router"},
})
try:
prompt = "Find the best Gmail tool for reading unread messages and explain the next required action."
recalled = await asyncio.to_thread(session.memory.context, {"query": prompt})
memory_context = recalled.get("promptText", "")
stacklink_tools = create_stacklink_claude_agent_tool_config(session)
options = ClaudeAgentOptions(
model=require_environment("CLAUDE_MODEL"),
system_prompt=(
"Use Stacklink search, exact contracts, connection management, "
"and governed execution to complete the request. Keep the same retry key for the same action. If an action is pending, inspect that operation; never submit it again with a new key."
+ "\n\n" + memory_context
),
tools=[],
max_turns=6,
mcp_servers=stacklink_tools["mcp_servers"],
allowed_tools=stacklink_tools["allowed_tools"],
)
final_output = ""
async for message in query(
prompt=prompt,
options=options,
):
if isinstance(message, ResultMessage) and not message.is_error:
final_output = message.result or ""
if bool(final_output):
await asyncio.to_thread(session.memory.record_conversation_turn, {
"applicationKey": "your-app",
"turnId": str(uuid.uuid4()),
"content": json.dumps({
"schemaVersion": "stacklink.conversation.turn.v1",
"user": {"role": "user", "content": prompt},
"assistant": {"role": "assistant", "content": final_output},
}, ensure_ascii=False),
})
print(final_output)
finally:
# Keep the application session while an accepted action is still pending.
cursor = None
while True:
page = stacklink.executions.list(session.id, cursor=cursor, limit=100)
if page["liveCount"]:
print("Keep this session for the pending operation:", session.id)
break
cursor = page["nextCursor"]
if not cursor:
session.close()
break
def require_environment(name: str) -> str:
value = os.getenv(name)
if not value:
raise RuntimeError(f"{name} is required.")
return value
if __name__ == "__main__":
asyncio.run(main())Shared approvals
Available from SDK 0.3.6. Bots, Studio, Workbench and direct tool calls use the same stored approval. An approval decision does not submit the action again.
Keep the project key on your server. Listing and observation require platform_approvals:read; decisions and customer-grant management require platform_approvals:manage. A key represents your application, not a company approver. Customer decisions additionally require a verified customer session and an active grant covering the exact action, account and resources.
- Retrieve the request and show its exact action and values. Use the returned
permissionsto enable decisions. - After the authorized customer explicitly chooses, submit that choice with the revision they saw. Do not approve automatically when an agent encounters a wait.
- Continue observing the original execution. A customer confirmation alone does not satisfy a separate company approval.
import os
from stacklink import Stacklink
stacklink = Stacklink(api_key=os.environ["STACKLINK_API_KEY"])
# Call from your authenticated server handler after the customer chooses.
# Resolve session_id from your signed-in customer, never from an arbitrary ID.
def submit_customer_decision(session_id, approval_id, displayed_revision, decision):
current = stacklink.approvals.get(
approval_id, session_id=session_id, as_customer=True,
)
permissions = current["permissions"]
if not permissions["canDecide"] or permissions["requirement"] != "customer":
raise PermissionError("This customer cannot decide this request.")
return stacklink.approvals.decide(
approval_id,
{"requirement": "customer", "decision": decision,
"expectedRevision": displayed_revision},
session_id=session_id, as_customer=True,
)Use approvals.list(session_id=session_id, as_customer=True) for the current customer. approvals.observe(session_id=session_id, as_customer=True, after=cursor) yields ordered updates; save each event's sequence and pass it as after after reconnecting. Use get_customer_grant, save_customer_grant and remove_customer_grant only from your authorized server.
A revision conflict means the displayed proposal is stale: retrieve and display the current request before asking for another decision. A failed notification is separate from a failed action. A final approval decision may still be followed by execution or an execution failure.
Source of truth
These examples are generated from the same executable files used by Stacklink Overview and release certification. npm and PyPI deliver the packages; this documentation remains the onboarding source of truth.