diff --git a/apps/sim/app/api/tools/managed-agent/list/route.ts b/apps/sim/app/api/tools/managed-agent/list/route.ts new file mode 100644 index 00000000000..1776f198793 --- /dev/null +++ b/apps/sim/app/api/tools/managed-agent/list/route.ts @@ -0,0 +1,146 @@ +import { AuditAction, AuditResourceType, recordAudit } from '@sim/audit' +import { createLogger } from '@sim/logger' +import { getErrorMessage } from '@sim/utils/errors' +import { type NextRequest, NextResponse } from 'next/server' +import { + listManagedAgentOptionsContract, + type ManagedAgentOption, + type ManagedAgentResource, +} from '@/lib/api/contracts/managed-agents' +import { parseRequest } from '@/lib/api/server' +import { authorizeCredentialUse } from '@/lib/auth/credential-access' +import { withRouteHandler } from '@/lib/core/utils/with-route-handler' +import { CLAUDE_PLATFORM_SERVICE_ACCOUNT_PROVIDER_ID } from '@/lib/credentials/token-service-accounts/descriptors' +import { AGENT_MEMORY_BETA, managedAgentsList } from '@/lib/managed-agents/session-client' +import { captureServerEvent } from '@/lib/posthog/server' +import { resolveOAuthAccountId, resolveServiceAccountToken } from '@/app/api/auth/oauth/utils' + +export const dynamic = 'force-dynamic' + +const logger = createLogger('ManagedAgentListAPI') + +interface AnthropicListRow { + id?: string + name?: string | null + config?: { type?: 'cloud' | 'self_hosted' } +} + +/** + * Anthropic list path + the beta header it requires. Memory-store endpoints + * use `agent-memory-2026-07-22`; combining it with the managed-agents beta is + * a documented 400, so each resource declares exactly one. + */ +const RESOURCE_ENDPOINTS: Record = { + agents: { path: '/v1/agents' }, + environments: { path: '/v1/environments' }, + vaults: { path: '/v1/vaults' }, + 'memory-stores': { path: '/v1/memory_stores', beta: AGENT_MEMORY_BETA }, +} + +function toOption( + resource: ManagedAgentResource, + row: AnthropicListRow +): ManagedAgentOption | null { + if (!row.id) return null + const name = row.name?.trim() + if (resource === 'environments') { + const type = row.config?.type + const suffix = type ? ` (${type})` : '' + return { id: row.id, label: `${name || row.id}${suffix}`, ...(type ? { type } : {}) } + } + if (resource === 'vaults') { + return { id: row.id, label: name || row.id } + } + return { id: row.id, label: name ? `${name} (${row.id})` : row.id } +} + +/** + * Resolves Managed Agent dropdown options (agents / environments / vaults / + * memory stores) for the block editor against a selected Claude Platform + * credential. The credential's API key is decrypted server-side and never + * crosses the client boundary — the browser only ever receives `{ id, label }` + * options. + */ +export const GET = withRouteHandler(async (request: NextRequest) => { + const parsed = await parseRequest(listManagedAgentOptionsContract, request, {}) + if (!parsed.success) return parsed.response + const { credentialId, resource } = parsed.data.query + + // Authenticates the caller AND verifies they may use this credential. + const authz = await authorizeCredentialUse(request, { + credentialId, + requireWorkflowIdForInternal: false, + }) + if (!authz.ok) { + return NextResponse.json({ error: authz.error || 'Unauthorized' }, { status: 403 }) + } + + const resolved = await resolveOAuthAccountId(credentialId) + if ( + resolved?.credentialType !== 'service_account' || + resolved.providerId !== CLAUDE_PLATFORM_SERVICE_ACCOUNT_PROVIDER_ID + ) { + return NextResponse.json({ error: 'Not a Claude Platform credential' }, { status: 400 }) + } + + let apiKey: string + try { + const token = await resolveServiceAccountToken( + credentialId, + CLAUDE_PLATFORM_SERVICE_ACCOUNT_PROVIDER_ID + ) + apiKey = token.accessToken + } catch (error) { + logger.warn('Failed to resolve Claude Platform credential', { error: getErrorMessage(error) }) + return NextResponse.json({ options: [] }) + } + + // Decrypting and using the credential's key is a credential access — record + // it, mirroring the OAuth token route's service-account audit trail. + const actorId = authz.requesterUserId + const workspaceId = resolved.workspaceId ?? authz.workspaceId ?? null + if (actorId) { + recordAudit({ + workspaceId, + actorId, + action: AuditAction.CREDENTIAL_ACCESSED, + resourceType: AuditResourceType.CREDENTIAL, + resourceId: credentialId, + description: 'Accessed Claude Platform credential to list Managed Agent resources', + metadata: { + provider: CLAUDE_PLATFORM_SERVICE_ACCOUNT_PROVIDER_ID, + credentialType: 'service_account', + }, + request, + }) + captureServerEvent( + actorId, + 'credential_used', + { + credential_type: 'service_account', + provider_id: CLAUDE_PLATFORM_SERVICE_ACCOUNT_PROVIDER_ID, + ...(workspaceId ? { workspace_id: workspaceId } : {}), + }, + workspaceId ? { groups: { workspace: workspaceId } } : undefined + ) + } + + try { + const endpoint = RESOURCE_ENDPOINTS[resource] + const rows = await managedAgentsList({ + apiKey, + path: endpoint.path, + beta: endpoint.beta, + signal: request.signal, + }) + const options = rows + .map((row) => toOption(resource, row)) + .filter((option): option is ManagedAgentOption => option !== null) + return NextResponse.json({ options }) + } catch (error) { + // Some beta workspaces may not expose every resource (e.g. vaults). Log + // and degrade to an empty list rather than breaking the editor. + logger.warn('Managed agent list proxy failed', { resource, error: getErrorMessage(error) }) + return NextResponse.json({ options: [] }) + } +}) diff --git a/apps/sim/app/workspace/[workspaceId]/integrations/components/connect-service-account-modal/connect-service-account-modal.tsx b/apps/sim/app/workspace/[workspaceId]/integrations/components/connect-service-account-modal/connect-service-account-modal.tsx index 83ae7ad48b8..016455bf012 100644 --- a/apps/sim/app/workspace/[workspaceId]/integrations/components/connect-service-account-modal/connect-service-account-modal.tsx +++ b/apps/sim/app/workspace/[workspaceId]/integrations/components/connect-service-account-modal/connect-service-account-modal.tsx @@ -108,6 +108,8 @@ interface ConnectServiceAccountModalProps { credentialDisplayName?: string /** Existing description, used to seed reconnect-capable modals. */ credentialDescription?: string + /** Called with the new credential id after a successful create (token-paste providers). */ + onCreated?: (credentialId: string) => void } /** @@ -132,6 +134,7 @@ export function ConnectServiceAccountModal({ credentialId, credentialDisplayName, credentialDescription, + onCreated, }: ConnectServiceAccountModalProps) { const clientCredentialDescriptor = getClientCredentialAccountDescriptor(serviceAccountProviderId) if (clientCredentialDescriptor) { @@ -162,6 +165,7 @@ export function ConnectServiceAccountModal({ credentialId={credentialId} initialDisplayName={credentialDisplayName} initialDescription={credentialDescription} + onCreated={onCreated} /> ) } diff --git a/apps/sim/app/workspace/[workspaceId]/integrations/components/connect-service-account-modal/token-service-account-modal.tsx b/apps/sim/app/workspace/[workspaceId]/integrations/components/connect-service-account-modal/token-service-account-modal.tsx index 186025caf52..ae2bdc00368 100644 --- a/apps/sim/app/workspace/[workspaceId]/integrations/components/connect-service-account-modal/token-service-account-modal.tsx +++ b/apps/sim/app/workspace/[workspaceId]/integrations/components/connect-service-account-modal/token-service-account-modal.tsx @@ -72,6 +72,8 @@ interface TokenServiceAccountModalProps { credentialId?: string initialDisplayName?: string initialDescription?: string + /** Called with the new credential id after a successful create (not reconnect). */ + onCreated?: (credentialId: string) => void } /** @@ -91,6 +93,7 @@ export function TokenServiceAccountModal({ credentialId, initialDisplayName, initialDescription, + onCreated, }: TokenServiceAccountModalProps) { const [apiToken, setApiToken] = useState('') const [domain, setDomain] = useState('') @@ -139,7 +142,7 @@ export function TokenServiceAccountModal({ description: description.trim() || undefined, }) } else { - await createCredential.mutateAsync({ + const created = await createCredential.mutateAsync({ workspaceId, type: 'service_account', providerId: descriptor.providerId, @@ -147,6 +150,7 @@ export function TokenServiceAccountModal({ displayName: displayName.trim() || undefined, description: description.trim() || undefined, }) + onCreated?.(created.credential.id) } onOpenChange(false) } catch (err: unknown) { diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/credential-selector/credential-selector.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/credential-selector/credential-selector.tsx index 51ddb37ef90..c30f39fc074 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/credential-selector/credential-selector.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/credential-selector/credential-selector.tsx @@ -12,8 +12,12 @@ import { type OAuthProvider, parseProvider, } from '@/lib/oauth' -import { getMissingRequiredScopes } from '@/lib/oauth/utils' +import { getMissingRequiredScopes, getServiceConfigByServiceId } from '@/lib/oauth/utils' import { ConnectOAuthModal } from '@/app/workspace/[workspaceId]/components/connect-oauth-modal' +import { + ConnectServiceAccountModal, + type ServiceAccountProviderId, +} from '@/app/workspace/[workspaceId]/integrations/components/connect-service-account-modal' import { ConnectSlackBotModal } from '@/app/workspace/[workspaceId]/integrations/components/connect-slack-bot-modal/connect-slack-bot-modal' import { formatDisplayText } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/formatted-text' import { getWorkflowSearchLabelHighlight } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/workflow-search-highlight' @@ -50,6 +54,7 @@ export function CredentialSelector({ const [showConnectModal, setShowConnectModal] = useState(false) const [showOAuthModal, setShowOAuthModal] = useState(false) const [showSlackBotModal, setShowSlackBotModal] = useState(false) + const [showServiceAccountModal, setShowServiceAccountModal] = useState(false) const [editingValue, setEditingValue] = useState('') const [isEditing, setIsEditing] = useState(false) const activeWorkflowId = useWorkflowRegistry((state) => state.activeWorkflowId) @@ -101,9 +106,9 @@ export function CredentialSelector({ const credentialKind = subBlock.credentialKind const credentials = useMemo(() => { - // A custom-bot picker lists only the reusable Slack bot credentials - // (service-account type), including in trigger mode. - if (credentialKind === 'custom-bot') { + // A custom-bot or service-account picker lists only the reusable + // service-account credentials, including in trigger mode. + if (credentialKind === 'custom-bot' || credentialKind === 'service-account') { return rawCredentials.filter((cred) => cred.type === 'service_account') } return isTriggerMode && !subBlock.allowServiceAccounts @@ -111,6 +116,14 @@ export function CredentialSelector({ : rawCredentials }, [rawCredentials, isTriggerMode, credentialKind, subBlock.allowServiceAccounts]) + // Resolved service-account provider metadata for the token-paste connect + // modal. Gated on `credentialKind` and using the non-throwing lookup so it + // never runs (or throws) for the OAuth / custom-bot pickers. + const serviceAccountService = useMemo( + () => (credentialKind === 'service-account' ? getServiceConfigByServiceId(serviceId) : null), + [credentialKind, serviceId] + ) + const selectedCredential = useMemo( () => credentials.find((cred) => cred.id === selectedId), [credentials, selectedId] @@ -189,6 +202,10 @@ export function CredentialSelector({ setShowSlackBotModal(true) return } + if (credentialKind === 'service-account') { + setShowServiceAccountModal(true) + return + } setShowConnectModal(true) }, [credentialKind]) @@ -235,9 +252,13 @@ export function CredentialSelector({ ? credentials.length > 0 ? 'Connect another custom bot' : 'Set up a custom bot' - : credentials.length > 0 - ? `Connect another ${getProviderName(provider)} account` - : `Connect ${getProviderName(provider)} account`, + : credentialKind === 'service-account' + ? credentials.length > 0 + ? `Add another ${getProviderName(provider)} key` + : `Add ${getProviderName(provider)} key` + : credentials.length > 0 + ? `Connect another ${getProviderName(provider)} account` + : `Connect ${getProviderName(provider)} account`, value: '__connect_account__', iconElement: , }) @@ -407,6 +428,23 @@ export function CredentialSelector({ }} /> )} + + {showServiceAccountModal && serviceAccountService?.serviceAccountProviderId && ( + { + setStoreValue(newCredentialId) + refetchCredentials() + }} + /> + )} ) } diff --git a/apps/sim/blocks/blocks/managed_agent.ts b/apps/sim/blocks/blocks/managed_agent.ts new file mode 100644 index 00000000000..894fdd4bc16 --- /dev/null +++ b/apps/sim/blocks/blocks/managed_agent.ts @@ -0,0 +1,230 @@ +import { ClaudeIcon } from '@/components/icons' +import { + fetchManagedAgentAgentOptions, + fetchManagedAgentEnvironmentOptions, + fetchManagedAgentMemoryStoreOptions, + fetchManagedAgentVaultOptions, +} from '@/lib/managed-agents/subblock-options' +import type { BlockConfig, BlockMeta } from '@/blocks/types' +import { AuthMode, IntegrationType } from '@/blocks/types' + +/** + * Claude Managed Agents block. + * + * Invokes a Claude Platform Managed Agent (cloud or self-hosted) as a workflow + * node and returns the assistant's final text. One block covers both models: + * the Environment type selector filters the environment list and shows only the + * fields that apply — memory stores and files are cloud-only (self-hosted + * rejects the `resources` attach), while session metadata works for both. + * + * Authentication is a selectable Claude Platform credential (an Anthropic + * workspace API key). The credential's key is resolved server-side at run + * time and never enters the block config or the browser. + */ +export const ManagedAgentBlock: BlockConfig = { + type: 'managed_agent', + name: 'Claude Managed Agents', + description: 'Run a Claude Platform Managed Agent', + authMode: AuthMode.ApiKey, + longDescription: + "Invoke a Claude Platform Managed Agent from a workflow. Select a Claude Platform account, pick an agent and environment from that workspace, optionally attach vaults, a memory store, and files, and add metadata tags. Returns the assistant's final text.", + category: 'tools', + integrationType: IntegrationType.AI, + docsLink: 'https://docs.sim.ai/integrations/managed-agent', + bgColor: '#DA7756', + iconColor: '#DA7756', + icon: ClaudeIcon, + subBlocks: [ + { + id: 'credential', + title: 'Claude Platform account', + type: 'oauth-input', + serviceId: 'claude-platform', + credentialKind: 'service-account', + required: true, + placeholder: 'Select a Claude Platform credential', + }, + { + id: 'environmentType', + title: 'Environment type', + type: 'dropdown', + required: true, + options: [ + { label: 'Cloud', id: 'cloud' }, + { label: 'Self-hosted', id: 'self_hosted' }, + ], + value: () => 'cloud', + description: + 'Self-hosted environments run on your own infrastructure and route memory via session metadata; file attachments are cloud-only.', + }, + { + id: 'agent', + title: 'Agent', + type: 'combobox', + required: true, + placeholder: 'Select an agent from your Claude workspace…', + commandSearchable: true, + options: [], + dependsOn: ['credential'], + fetchOptions: fetchManagedAgentAgentOptions, + }, + { + id: 'environment', + title: 'Environment', + type: 'combobox', + required: true, + placeholder: 'Select an environment…', + commandSearchable: true, + options: [], + dependsOn: ['credential', 'environmentType'], + fetchOptions: fetchManagedAgentEnvironmentOptions, + }, + { + id: 'userMessage', + title: 'User message', + type: 'long-input', + required: true, + placeholder: 'Ask the Managed Agent to do something…', + }, + { + id: 'vaults', + title: 'Credential vaults', + type: 'dropdown', + required: false, + mode: 'advanced', + placeholder: 'Optional — pick zero or more OAuth vaults', + searchable: true, + multiSelect: true, + options: [], + dependsOn: ['credential'], + fetchOptions: fetchManagedAgentVaultOptions, + }, + { + id: 'vaultsAck', + title: + 'I own or am authorized to use these vaults. I understand this means this agent can assume the identity granted by them.', + type: 'switch', + required: false, + mode: 'advanced', + description: 'Required when at least one vault is selected above.', + }, + { + id: 'memoryStoreId', + title: 'Memory store', + type: 'combobox', + required: false, + mode: 'advanced', + placeholder: 'Optional — pick a memory store', + commandSearchable: true, + options: [], + dependsOn: ['credential'], + // Cloud only: memory stores attach as `resources[]`, which self-hosted + // rejects. A self-hosted worker that uses a store reads its id from a + // Metadata key the author sets explicitly. + condition: { field: 'environmentType', value: 'cloud' }, + fetchOptions: fetchManagedAgentMemoryStoreOptions, + }, + { + id: 'memoryAccess', + title: 'Memory access', + type: 'dropdown', + required: false, + mode: 'advanced', + options: [ + { label: 'Read + write (default)', id: 'read_write' }, + { label: 'Read only', id: 'read_only' }, + ], + value: () => 'read_write', + condition: { + field: 'memoryStoreId', + value: '', + not: true, + and: { field: 'environmentType', value: 'cloud' }, + }, + description: 'read_write pushes changes back on session exit; read_only never writes.', + }, + { + id: 'memoryInstructions', + title: 'Memory instructions', + type: 'long-input', + required: false, + mode: 'advanced', + placeholder: 'Optional — how the agent should use this memory store', + // Cloud only: instructions are a `resources[]` memory-attach concept the + // API renders into the system prompt; self-hosted has no resource attach. + condition: { + field: 'memoryStoreId', + value: '', + not: true, + and: { field: 'environmentType', value: 'cloud' }, + }, + description: 'Per-attachment guidance rendered into the memory section of the system prompt.', + }, + { + id: 'files', + title: 'Files', + type: 'table', + required: false, + mode: 'advanced', + // Cloud only: files attach as `resources[]`, which self-hosted rejects. + condition: { field: 'environmentType', value: 'cloud' }, + columns: ['File ID', 'Mount path'], + description: + 'Files-API file ids (file_...) to attach as file resources. Mount path is optional.', + }, + { + id: 'sessionParameters', + title: 'Metadata', + type: 'table', + required: false, + mode: 'advanced', + columns: ['Key', 'Value'], + description: + 'Optional key/value metadata forwarded on the session. On self-hosted environments each key is exposed to the agent as an env var.', + }, + ], + tools: { + access: ['managed_agent_run_session'], + }, + inputs: { + credential: { type: 'string', description: 'Claude Platform credential id.' }, + environmentType: { + type: 'string', + description: + "Environment execution model — 'cloud' or 'self_hosted'. Filters the environment picker and gates cloud-only fields; the actual type is re-resolved server-side for routing.", + }, + agent: { type: 'string', description: 'Managed-agent id inside the linked Claude workspace.' }, + environment: { + type: 'string', + description: 'Environment id inside the linked Claude workspace.', + }, + userMessage: { type: 'string', description: 'The user message to send to the agent.' }, + vaults: { type: 'json', description: 'Vault ids for MCP auth (array of strings).' }, + vaultsAck: { + type: 'boolean', + description: 'Acknowledgement that the author may use the attached vaults.', + }, + memoryStoreId: { type: 'string', description: 'Optional Agent Memory Store id.' }, + memoryAccess: { + type: 'string', + description: "Memory store access mode — 'read_write' (default) or 'read_only'.", + }, + memoryInstructions: { + type: 'string', + description: 'Per-attachment guidance for how the agent should use the memory store.', + }, + files: { type: 'json', description: 'File attachments — [{fileId, mountPath?}].' }, + sessionParameters: { type: 'json', description: 'Session metadata (key/value).' }, + }, + outputs: { + content: { type: 'string', description: "The Managed Agent's final assistant text." }, + sessionId: { type: 'string', description: 'Anthropic session id, for logs and linking.' }, + inputTokens: { type: 'number', description: 'Cumulative input tokens for the session.' }, + outputTokens: { type: 'number', description: 'Cumulative output tokens for the session.' }, + }, +} + +export const ManagedAgentBlockMeta = { + tags: ['agentic', 'llm'], + url: 'https://platform.claude.com/', +} as const satisfies BlockMeta diff --git a/apps/sim/blocks/registry-maps.ts b/apps/sim/blocks/registry-maps.ts index 65a48b4c310..d3bc8cea720 100644 --- a/apps/sim/blocks/registry-maps.ts +++ b/apps/sim/blocks/registry-maps.ts @@ -183,6 +183,7 @@ import { LoopsBlock, LoopsBlockMeta } from '@/blocks/blocks/loops' import { LumaBlock, LumaBlockMeta } from '@/blocks/blocks/luma' import { MailchimpBlock, MailchimpBlockMeta } from '@/blocks/blocks/mailchimp' import { MailgunBlock, MailgunBlockMeta } from '@/blocks/blocks/mailgun' +import { ManagedAgentBlock, ManagedAgentBlockMeta } from '@/blocks/blocks/managed_agent' import { ManualTriggerBlock } from '@/blocks/blocks/manual_trigger' import { McpBlock } from '@/blocks/blocks/mcp' import { Mem0Block, Mem0BlockMeta } from '@/blocks/blocks/mem0' @@ -510,6 +511,7 @@ export const BLOCK_REGISTRY: Record = { luma: LumaBlock, mailchimp: MailchimpBlock, mailgun: MailgunBlock, + managed_agent: ManagedAgentBlock, manual_trigger: ManualTriggerBlock, mcp: McpBlock, mem0: Mem0Block, @@ -807,6 +809,7 @@ export const BLOCK_META_REGISTRY: Record = { luma: LumaBlockMeta, mailchimp: MailchimpBlockMeta, mailgun: MailgunBlockMeta, + managed_agent: ManagedAgentBlockMeta, mem0: Mem0BlockMeta, microsoft_ad: MicrosoftAdBlockMeta, microsoft_dataverse: MicrosoftDataverseBlockMeta, diff --git a/apps/sim/blocks/types.ts b/apps/sim/blocks/types.ts index 9024e44cb88..43b401db84c 100644 --- a/apps/sim/blocks/types.ts +++ b/apps/sim/blocks/types.ts @@ -370,8 +370,11 @@ export interface SubBlockConfig { * Narrows an `oauth-input` selector to a specific credential kind. `'custom-bot'` * lists only reusable custom Slack bot credentials (service-account type) and its * connect row opens the custom-bot setup modal instead of the OAuth flow. + * `'service-account'` is the generic equivalent for a no-OAuth provider: it lists + * only service-account credentials and its connect row opens the descriptor-driven + * token-paste modal (`ConnectServiceAccountModal`). */ - credentialKind?: 'custom-bot' + credentialKind?: 'custom-bot' | 'service-account' /** * Opts a trigger-mode `oauth-input` selector into listing service-account * credentials, which are otherwise excluded in trigger mode. Set only when the diff --git a/apps/sim/components/icons.tsx b/apps/sim/components/icons.tsx index 9ac49e147a7..94aad863fe6 100644 --- a/apps/sim/components/icons.tsx +++ b/apps/sim/components/icons.tsx @@ -3683,6 +3683,20 @@ export const AnthropicIcon = (props: SVGProps) => ( ) +export const ClaudeIcon = (props: SVGProps) => ( + + Claude + + +) + export function AzureIcon(props: SVGProps) { const id = useId() const gradient0 = `azure_paint0_${id}` diff --git a/apps/sim/lib/api/contracts/index.ts b/apps/sim/lib/api/contracts/index.ts index ef339965c5b..259878e0935 100644 --- a/apps/sim/lib/api/contracts/index.ts +++ b/apps/sim/lib/api/contracts/index.ts @@ -13,6 +13,7 @@ export * from './file-uploads' export * from './folders' export * from './hotspots' export * from './inbox' +export * from './managed-agents' export * from './media' export * from './permission-groups' export * from './primitives' diff --git a/apps/sim/lib/api/contracts/managed-agents.ts b/apps/sim/lib/api/contracts/managed-agents.ts new file mode 100644 index 00000000000..eef276b3de4 --- /dev/null +++ b/apps/sim/lib/api/contracts/managed-agents.ts @@ -0,0 +1,42 @@ +import { z } from 'zod' +import { type ContractJsonResponse, defineRouteContract } from '@/lib/api/contracts/types' + +/** + * Contract for the Managed Agent block-editor dropdowns. The `list` route + * resolves agents / environments / vaults / memory stores against a selected + * Claude Platform credential, decrypting its API key server-side — the key + * never crosses the client boundary. + */ + +/** A dropdown option resolved from the linked Claude Platform workspace. */ +export const managedAgentOptionSchema = z.object({ + id: z.string(), + label: z.string(), + /** Environment execution model — only set for the `environments` resource, used to filter by mode. */ + type: z.enum(['cloud', 'self_hosted']).optional(), +}) +export type ManagedAgentOption = z.output + +export const managedAgentResourceSchema = z.enum([ + 'agents', + 'environments', + 'vaults', + 'memory-stores', +]) +export type ManagedAgentResource = z.output + +export const listManagedAgentOptionsQuerySchema = z.object({ + credentialId: z.string().min(1, 'A Claude Platform credential is required'), + resource: managedAgentResourceSchema, +}) + +export const listManagedAgentOptionsContract = defineRouteContract({ + method: 'GET', + path: '/api/tools/managed-agent/list', + query: listManagedAgentOptionsQuerySchema, + response: { + mode: 'json', + schema: z.object({ options: z.array(managedAgentOptionSchema) }), + }, +}) +export type ListManagedAgentOptions = ContractJsonResponse diff --git a/apps/sim/lib/credentials/token-service-accounts/descriptors.ts b/apps/sim/lib/credentials/token-service-accounts/descriptors.ts index a5939f10575..3b3de5551d0 100644 --- a/apps/sim/lib/credentials/token-service-accounts/descriptors.ts +++ b/apps/sim/lib/credentials/token-service-accounts/descriptors.ts @@ -69,6 +69,8 @@ export const TRELLO_SERVICE_ACCOUNT_PROVIDER_ID = 'trello-service-account' as co export const CALCOM_SERVICE_ACCOUNT_PROVIDER_ID = 'calcom-service-account' as const export const WEALTHBOX_SERVICE_ACCOUNT_PROVIDER_ID = 'wealthbox-service-account' as const export const PIPEDRIVE_SERVICE_ACCOUNT_PROVIDER_ID = 'pipedrive-service-account' as const +export const CLAUDE_PLATFORM_SERVICE_ACCOUNT_PROVIDER_ID = + 'claude-platform-service-account' as const const SHOPIFY_DOMAIN_HINT_REGEX = /^[a-z0-9][a-z0-9-]*\.myshopify\.com$/i @@ -87,6 +89,7 @@ export type TokenServiceAccountProviderId = | typeof CALCOM_SERVICE_ACCOUNT_PROVIDER_ID | typeof WEALTHBOX_SERVICE_ACCOUNT_PROVIDER_ID | typeof PIPEDRIVE_SERVICE_ACCOUNT_PROVIDER_ID + | typeof CLAUDE_PLATFORM_SERVICE_ACCOUNT_PROVIDER_ID export const TOKEN_SERVICE_ACCOUNT_DESCRIPTORS: Record< TokenServiceAccountProviderId, @@ -350,6 +353,25 @@ export const TOKEN_SERVICE_ACCOUNT_DESCRIPTORS: Record< 'Each Pipedrive user has one API token per company — regenerating it breaks every integration using the old value, and API-token traffic gets lower rate limits than OAuth.', authStyle: 'x-api-token', }, + [CLAUDE_PLATFORM_SERVICE_ACCOUNT_PROVIDER_ID]: { + providerId: CLAUDE_PLATFORM_SERVICE_ACCOUNT_PROVIDER_ID, + serviceLabel: 'Claude Platform', + tokenNoun: 'API key', + connectNoun: 'API key', + fields: [ + { + id: 'apiToken', + label: 'API key', + placeholder: 'sk-ant-...', + secret: true, + hintPattern: /^sk-ant-/, + hintMessage: 'Claude Platform API keys usually start with sk-ant-.', + }, + ], + docsUrl: 'https://docs.sim.ai/integrations/managed-agent', + helpText: + 'Use an Anthropic workspace API key with Managed Agents access. The key is scoped to one Anthropic workspace — its agents, environments, vaults, and memory stores.', + }, } /** diff --git a/apps/sim/lib/credentials/token-service-accounts/server.ts b/apps/sim/lib/credentials/token-service-accounts/server.ts index 9f01c3c0027..a7a693b1ca0 100644 --- a/apps/sim/lib/credentials/token-service-accounts/server.ts +++ b/apps/sim/lib/credentials/token-service-accounts/server.ts @@ -3,6 +3,7 @@ import { ASANA_SERVICE_ACCOUNT_PROVIDER_ID, ATTIO_SERVICE_ACCOUNT_PROVIDER_ID, CALCOM_SERVICE_ACCOUNT_PROVIDER_ID, + CLAUDE_PLATFORM_SERVICE_ACCOUNT_PROVIDER_ID, CLICKUP_SERVICE_ACCOUNT_PROVIDER_ID, HUBSPOT_SERVICE_ACCOUNT_PROVIDER_ID, isTokenServiceAccountProviderId, @@ -21,6 +22,7 @@ import { validateAirtableServiceAccount } from '@/lib/credentials/token-service- import { validateAsanaServiceAccount } from '@/lib/credentials/token-service-accounts/validators/asana' import { validateAttioServiceAccount } from '@/lib/credentials/token-service-accounts/validators/attio' import { validateCalcomServiceAccount } from '@/lib/credentials/token-service-accounts/validators/calcom' +import { validateClaudePlatformServiceAccount } from '@/lib/credentials/token-service-accounts/validators/claude-platform' import { validateClickupServiceAccount } from '@/lib/credentials/token-service-accounts/validators/clickup' import { validateHubspotServiceAccount } from '@/lib/credentials/token-service-accounts/validators/hubspot' import { validateLinearServiceAccount } from '@/lib/credentials/token-service-accounts/validators/linear' @@ -80,6 +82,7 @@ const TOKEN_SERVICE_ACCOUNT_VALIDATORS: Record< [CALCOM_SERVICE_ACCOUNT_PROVIDER_ID]: validateCalcomServiceAccount, [WEALTHBOX_SERVICE_ACCOUNT_PROVIDER_ID]: validateWealthboxServiceAccount, [PIPEDRIVE_SERVICE_ACCOUNT_PROVIDER_ID]: validatePipedriveServiceAccount, + [CLAUDE_PLATFORM_SERVICE_ACCOUNT_PROVIDER_ID]: validateClaudePlatformServiceAccount, } export function getTokenServiceAccountValidator( diff --git a/apps/sim/lib/credentials/token-service-accounts/validators/claude-platform.ts b/apps/sim/lib/credentials/token-service-accounts/validators/claude-platform.ts new file mode 100644 index 00000000000..f457d6be29d --- /dev/null +++ b/apps/sim/lib/credentials/token-service-accounts/validators/claude-platform.ts @@ -0,0 +1,42 @@ +import { + fetchProvider, + throwForProviderResponse, +} from '@/lib/credentials/token-service-accounts/errors' +import type { + TokenServiceAccountFields, + TokenServiceAccountValidationResult, +} from '@/lib/credentials/token-service-accounts/server' + +const AGENTS_URL = 'https://api.anthropic.com/v1/agents?limit=1' +const ANTHROPIC_VERSION = '2023-06-01' +const MANAGED_AGENTS_BETA = 'managed-agents-2026-04-01' + +/** + * Validates a Claude Platform API key by listing agents on the linked + * Anthropic workspace. 401/403 mean the key was rejected; any other non-2xx + * means Claude Platform is unavailable. The Managed Agents API has no + * "who am I" endpoint, so the key's last four characters are used as a + * human-distinguishable display name across multiple linked workspaces. + */ +export async function validateClaudePlatformServiceAccount( + fields: TokenServiceAccountFields +): Promise { + const res = await fetchProvider( + AGENTS_URL, + { + headers: { + 'x-api-key': fields.apiToken, + 'anthropic-version': ANTHROPIC_VERSION, + 'anthropic-beta': MANAGED_AGENTS_BETA, + }, + }, + 'agents_list' + ) + await throwForProviderResponse(res, 'agents_list') + + const suffix = fields.apiToken.slice(-4) + return { + displayName: `Claude Platform (…${suffix})`, + auditMetadata: {}, + } +} diff --git a/apps/sim/lib/managed-agents/run-session.test.ts b/apps/sim/lib/managed-agents/run-session.test.ts new file mode 100644 index 00000000000..52a5c66ba2b --- /dev/null +++ b/apps/sim/lib/managed-agents/run-session.test.ts @@ -0,0 +1,390 @@ +/** + * @vitest-environment node + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import type { AnthropicSessionEvent } from '@/lib/managed-agents/session-client' + +const { mocks } = vi.hoisted(() => ({ + mocks: { + createSession: vi.fn(), + sendUserMessage: vi.fn(), + sendSessionEvents: vi.fn(), + openSessionStream: vi.fn(), + listSessionEvents: vi.fn(), + getSession: vi.fn(), + getEnvironmentType: vi.fn(), + interruptSession: vi.fn(), + readSSEEvents: vi.fn(), + sleep: vi.fn(), + }, +})) + +vi.mock('@/lib/managed-agents/session-client', () => ({ + createSession: mocks.createSession, + sendUserMessage: mocks.sendUserMessage, + sendSessionEvents: mocks.sendSessionEvents, + openSessionStream: mocks.openSessionStream, + listSessionEvents: mocks.listSessionEvents, + getSession: mocks.getSession, + getEnvironmentType: mocks.getEnvironmentType, + interruptSession: mocks.interruptSession, +})) +vi.mock('@/lib/core/utils/sse', () => ({ readSSEEvents: mocks.readSSEEvents })) +vi.mock('@sim/utils/helpers', () => ({ sleep: mocks.sleep })) + +import { runManagedAgentSession } from '@/lib/managed-agents/run-session' + +/** Drives `readSSEEvents`: each call replays the next scripted batch of events. */ +function scriptStreamBatches(batches: AnthropicSessionEvent[][]): void { + let call = 0 + mocks.readSSEEvents.mockImplementation( + async (_resp: unknown, opts: { onEvent: (e: AnthropicSessionEvent) => Promise }) => { + const batch = batches[call++] ?? [] + for (const event of batch) { + const stop = await opts.onEvent(event) + if (stop === true) return + } + } + ) +} + +const BASE = { + apiKey: 'sk-ant-fake', + agentId: 'agent_1', + environmentId: 'env_1', + userMessage: 'do a thing', +} as const + +const msg = (id: string, text: string): AnthropicSessionEvent => ({ + id, + type: 'agent.message', + content: [{ type: 'text', text }], +}) +const idle = (id: string, stop: string): AnthropicSessionEvent => ({ + id, + type: 'session.status_idle', + stop_reason: { type: stop }, +}) + +beforeEach(() => { + vi.clearAllMocks() + mocks.createSession.mockResolvedValue({ id: 'sess_1' }) + mocks.sendUserMessage.mockResolvedValue(undefined) + mocks.sendSessionEvents.mockResolvedValue(undefined) + mocks.openSessionStream.mockResolvedValue({}) + mocks.listSessionEvents.mockResolvedValue([]) + mocks.getSession.mockResolvedValue(null) + mocks.getEnvironmentType.mockResolvedValue('cloud') + mocks.interruptSession.mockResolvedValue(undefined) +}) + +const customToolUse = (id: string, name: string): AnthropicSessionEvent => ({ + id, + type: 'agent.custom_tool_use', + name, +}) + +describe('runManagedAgentSession', () => { + it('accumulates agent.message text and completes on end_turn (terminal event)', async () => { + scriptStreamBatches([[msg('e1', 'Hello '), msg('e2', 'world'), idle('e3', 'end_turn')]]) + mocks.getSession.mockResolvedValue({ + status: 'idle', + usage: { inputTokens: 12, outputTokens: 3 }, + }) + + const result = await runManagedAgentSession({ ...BASE }) + + expect(result).toEqual({ + ok: true, + content: 'Hello world', + sessionId: 'sess_1', + inputTokens: 12, + outputTokens: 3, + }) + expect(mocks.listSessionEvents).not.toHaveBeenCalled() + }) + + it('completes via authoritative status when the stream goes quiet after progress', async () => { + // Stream: some text, no terminal, then closes. Reconnect: nothing new. + scriptStreamBatches([[msg('e1', 'partial')], []]) + // First getSession (quiet-reconnect check) → idle; final getSession → usage. + mocks.getSession + .mockResolvedValueOnce({ status: 'idle' }) + .mockResolvedValue({ status: 'idle', usage: { inputTokens: 5 } }) + + const result = await runManagedAgentSession({ ...BASE }) + + expect(result.ok).toBe(true) + expect(result.content).toBe('partial') + expect(result.inputTokens).toBe(5) + }) + + it('keeps waiting while status is running, then completes on a later terminal event', async () => { + // Stream 1: text, closes (no terminal). Reconnect: nothing new, status running → backoff. + // Stream 2: end_turn → complete. + scriptStreamBatches([[msg('e1', 'thinking')], [idle('e2', 'end_turn')]]) + mocks.getSession + .mockResolvedValueOnce({ status: 'running' }) + .mockResolvedValue({ status: 'idle' }) + + const result = await runManagedAgentSession({ ...BASE }) + + expect(result.ok).toBe(true) + expect(result.content).toBe('thinking') + expect(mocks.sleep).toHaveBeenCalled() // backed off while running + }) + + it('does not complete on idle status while a requires_action is outstanding', async () => { + // Stream 1: text then a requires_action idle (pending), then closes. + // Reconnect: nothing new, status idle — but must NOT complete (pending). + // Stream 2: end_turn → complete. + scriptStreamBatches([ + [msg('e1', 'partial'), idle('r1', 'requires_action')], + [idle('e2', 'end_turn')], + ]) + mocks.getSession.mockResolvedValue({ status: 'idle' }) + + const result = await runManagedAgentSession({ ...BASE }) + + expect(result.ok).toBe(true) + expect(result.content).toBe('partial') + expect(mocks.openSessionStream).toHaveBeenCalledTimes(2) // reopened rather than completing early + }) + + it('does not complete on a fresh idle before the agent has started', async () => { + // Stream closes immediately with nothing; catch-up empty; status idle but no + // activity yet → must NOT complete. Then it starts and finishes. + scriptStreamBatches([[], [idle('e1', 'end_turn')]]) + mocks.getSession + .mockResolvedValueOnce({ status: 'idle' }) // pre-start idle — must be ignored + .mockResolvedValue({ status: 'idle' }) + + const result = await runManagedAgentSession({ ...BASE }) + + expect(result.ok).toBe(true) + // Completed via the real end_turn on reopen, not the premature idle. + expect(mocks.openSessionStream).toHaveBeenCalledTimes(2) + }) + + it('surfaces a session.error as a failure with the error message', async () => { + scriptStreamBatches([[{ id: 'x1', type: 'session.error', error: { message: 'boom' } }]]) + + const result = await runManagedAgentSession({ ...BASE }) + + expect(result.ok).toBe(false) + expect(result.error).toBe('boom') + expect(result.sessionId).toBe('sess_1') + }) + + it('keeps requires_action pending when catch-up recovers an older message (ordering)', async () => { + // Live stream sees a message then a requires_action pause. Catch-up then + // recovers an EARLIER, unseen agent.message — processing it would clear the + // pending flag, but the history's latest lifecycle event is still + // requires_action, so the session must NOT be reported complete. + scriptStreamBatches([ + [msg('e1', 'hi '), idle('r1', 'requires_action')], + [idle('e2', 'end_turn')], + ]) + mocks.listSessionEvents.mockResolvedValueOnce([ + msg('e0', 'earlier'), // unseen, older than r1 + idle('r1', 'requires_action'), // latest lifecycle event in history + ]) + mocks.getSession.mockResolvedValue({ status: 'idle' }) + + const result = await runManagedAgentSession({ ...BASE }) + + expect(result.ok).toBe(true) + // Reopened rather than completing early on the idle snapshot. + expect(mocks.openSessionStream).toHaveBeenCalledTimes(2) + }) + + it('preserves live requires_action when catch-up has an older message but no lifecycle event', async () => { + // Live stream sees a message + requires_action pause. Catch-up recovers an + // older, unseen agent.message but NO lifecycle event — processing it clears + // the pending flag, and with no lifecycle evidence in history the live + // pending state must be restored (not completed on the idle snapshot). + scriptStreamBatches([ + [msg('e1', 'hi '), idle('r1', 'requires_action')], + [idle('e2', 'end_turn')], + ]) + mocks.listSessionEvents.mockResolvedValueOnce([msg('e0', 'earlier')]) // no lifecycle event + mocks.getSession.mockResolvedValue({ status: 'idle' }) + + const result = await runManagedAgentSession({ ...BASE }) + + expect(result.ok).toBe(true) + expect(mocks.openSessionStream).toHaveBeenCalledTimes(2) // did not complete on the idle snapshot + }) + + it('completes on an idless terminal event delivered only on the live stream', async () => { + // An idless session.status_idle(end_turn) must still register as terminal — + // idless events are processed (only text accumulation is id-gated). + scriptStreamBatches([[{ type: 'session.status_idle', stop_reason: { type: 'end_turn' } }]]) + mocks.getSession.mockResolvedValue({ status: 'idle' }) + + const result = await runManagedAgentSession({ ...BASE }) + + expect(result.ok).toBe(true) + expect(mocks.openSessionStream).toHaveBeenCalledTimes(1) // completed, not dropped + }) + + it('does not double-count text from an idless preview of a persisted message', async () => { + scriptStreamBatches([ + [ + { type: 'agent.message', content: [{ type: 'text', text: 'hi' }] }, // idless preview + msg('e1', 'hi'), // persisted copy of the same text + idle('e2', 'end_turn'), + ], + ]) + mocks.getSession.mockResolvedValue({ status: 'idle' }) + + const result = await runManagedAgentSession({ ...BASE }) + + expect(result.content).toBe('hi') // not 'hihi' + }) + + it('retries a custom-tool reply that failed to send instead of stranding the session', async () => { + // Stream 1: a custom tool call whose error reply fails to send — the event + // must stay unseen. Reconnect (status running) → reopen. Stream 2: the same + // tool call is retried (reply succeeds), then end_turn completes. + scriptStreamBatches([ + [customToolUse('t1', 'foo')], + [customToolUse('t1', 'foo'), idle('e2', 'end_turn')], + ]) + mocks.sendSessionEvents.mockRejectedValueOnce(new Error('network')).mockResolvedValue(undefined) + mocks.getSession + .mockResolvedValueOnce({ status: 'running' }) + .mockResolvedValue({ status: 'idle' }) + + const result = await runManagedAgentSession({ ...BASE }) + + expect(result.ok).toBe(true) + // Reply attempted twice: the failed send was retried on reconnect. + expect(mocks.sendSessionEvents).toHaveBeenCalledTimes(2) + }) + + it('keeps requires_action pending when lagging history holds only an older lifecycle event', async () => { + // Live sees requires_action@:02 (newest). Catch-up history LAGS — it holds + // only an older, unseen running@:00. By timestamp that older event must not + // clear the newer pause, so the idle snapshot must not complete the run. + const runningAt = (id: string, at: string): AnthropicSessionEvent => ({ + id, + type: 'session.status_running', + processed_at: at, + }) + const idleAt = (id: string, stop: string, at: string): AnthropicSessionEvent => ({ + id, + type: 'session.status_idle', + stop_reason: { type: stop }, + processed_at: at, + }) + scriptStreamBatches([ + [msg('e1', 'hi'), idleAt('ra1', 'requires_action', '2026-01-01T00:00:02Z')], + [idleAt('e2', 'end_turn', '2026-01-01T00:00:05Z')], + ]) + mocks.listSessionEvents.mockResolvedValueOnce([runningAt('run0', '2026-01-01T00:00:00Z')]) + mocks.getSession.mockResolvedValue({ status: 'idle' }) + + const result = await runManagedAgentSession({ ...BASE }) + + expect(result.ok).toBe(true) + expect(mocks.openSessionStream).toHaveBeenCalledTimes(2) // older running did not clear the newer pause + }) + + it('an untimestamped running event does not clear a timestamped requires_action', async () => { + // A stray lifecycle event without processed_at must not outrank (clear) a + // timestamped requires_action pause, nor poison later timestamped events. + const idleAt = (id: string, stop: string, at: string): AnthropicSessionEvent => ({ + id, + type: 'session.status_idle', + stop_reason: { type: stop }, + processed_at: at, + }) + scriptStreamBatches([ + [ + msg('e1', 'hi'), + idleAt('ra1', 'requires_action', '2026-01-01T00:00:02Z'), + { id: 'runX', type: 'session.status_running' }, // no processed_at + ], + [idleAt('e2', 'end_turn', '2026-01-01T00:00:05Z')], + ]) + mocks.getSession.mockResolvedValue({ status: 'idle' }) + + const result = await runManagedAgentSession({ ...BASE }) + + expect(result.ok).toBe(true) + expect(mocks.openSessionStream).toHaveBeenCalledTimes(2) // stray running did not clear the pause + }) + + it('does not complete on a session_idle event that carries no stop_reason', async () => { + // An idle event with no stop_reason (e.g. pre-first-turn) must NOT be + // treated as complete via the event path — defer to the status gate, which + // honors sawActivity. Then a real end_turn on reopen completes it. + scriptStreamBatches([ + [{ id: 'i1', type: 'session.status_idle' }], + [msg('e2', 'done'), idle('e3', 'end_turn')], + ]) + mocks.getSession.mockResolvedValue({ status: 'idle' }) + + const result = await runManagedAgentSession({ ...BASE }) + + expect(result.ok).toBe(true) + expect(result.content).toBe('done') + expect(mocks.openSessionStream).toHaveBeenCalledTimes(2) // reopened, not completed on i1 + }) + + it('backs off (does not hot-loop) when catch-up only surfaces a failing tool reply', async () => { + // Stream closes empty; catch-up surfaces an unseen custom-tool call whose + // reply fails to send — it stays unseen (retry), so it must NOT count as + // progress and reset the backoff. Session still running → we must sleep. + scriptStreamBatches([[], [idle('e2', 'end_turn')]]) + mocks.listSessionEvents + .mockResolvedValueOnce([customToolUse('t1', 'foo')]) + .mockResolvedValue([]) + mocks.sendSessionEvents.mockRejectedValue(new Error('network')) + mocks.getSession + .mockResolvedValueOnce({ status: 'running' }) + .mockResolvedValue({ status: 'idle' }) + + const result = await runManagedAgentSession({ ...BASE }) + + expect(result.ok).toBe(true) + expect(mocks.sleep).toHaveBeenCalled() // backed off instead of resetting on the retry event + }) + + it('interrupts the session when a mid-run stream failure ends the loop', async () => { + mocks.openSessionStream.mockRejectedValue(new Error('stream 502')) + + const result = await runManagedAgentSession({ ...BASE }) + + expect(result.ok).toBe(false) + expect(mocks.interruptSession).toHaveBeenCalledWith({ + apiKey: BASE.apiKey, + sessionId: 'sess_1', + }) + }) + + it('interrupts the session and reports aborted when the workflow is cancelled', async () => { + const controller = new AbortController() + // Cancel right after the session is created, before the stream loop runs. + mocks.sendUserMessage.mockImplementation(async () => { + controller.abort() + }) + scriptStreamBatches([[]]) + + const result = await runManagedAgentSession({ ...BASE, signal: controller.signal }) + + expect(result.ok).toBe(false) + expect(result.error).toBe('aborted') + expect(mocks.interruptSession).toHaveBeenCalledWith({ + apiKey: BASE.apiKey, + sessionId: 'sess_1', + }) + }) + + it('rejects an empty user message before creating a session', async () => { + const result = await runManagedAgentSession({ ...BASE, userMessage: ' ' }) + expect(result.ok).toBe(false) + expect(mocks.createSession).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/managed-agents/run-session.ts b/apps/sim/lib/managed-agents/run-session.ts new file mode 100644 index 00000000000..0fe2ea472dc --- /dev/null +++ b/apps/sim/lib/managed-agents/run-session.ts @@ -0,0 +1,434 @@ +import { createLogger } from '@sim/logger' +import { getErrorMessage } from '@sim/utils/errors' +import { sleep } from '@sim/utils/helpers' +import { readSSEEvents } from '@/lib/core/utils/sse' +import { + type AnthropicSessionEvent, + type CreateSessionInput, + createSession, + type EnvironmentType, + getEnvironmentType, + getSession, + interruptSession, + listSessionEvents, + openSessionStream, + sendSessionEvents, + sendUserMessage, +} from '@/lib/managed-agents/session-client' + +/** + * Runs a Claude Platform Managed Agent session end-to-end and returns the + * accumulated assistant text. This is the one genuinely-custom piece of the + * integration: the Managed Agents lifecycle is create → send → stream → + * catch-up → reconnect, which the single-request tool framework can't model. + * + * Completion is driven only by real terminal signals — a `session.status_idle` + * (`end_turn`) event, or the authoritative session `status` when the event + * stream is quiet — never by a heuristic timer. Pure with respect to Sim (no + * `@sim/db`, no executor types); the caller supplies the decrypted API key. + */ + +const logger = createLogger('ManagedAgentRunSession') + +/** Upper bound on stream-close → catch-up → reopen cycles per invocation. */ +const MAX_RECONNECT_ITERATIONS = 120 +/** Wall-clock backstop for the reconnect loop (not the live-stream duration). */ +const MAX_SESSION_MS = 15 * 60 * 1000 +const RECONNECT_BACKOFF_START_MS = 500 +const RECONNECT_BACKOFF_MAX_MS = 5000 + +export interface RunManagedAgentInput { + apiKey: string + agentId: string + environmentId: string + /** Env-type hint from the block; used only if server-side resolution fails. */ + environmentType?: EnvironmentType + userMessage: string + title?: string + vaultIds?: string[] + memoryStoreId?: string + memoryAccess?: 'read_write' | 'read_only' + memoryInstructions?: string + files?: Array<{ fileId: string; mountPath?: string }> + sessionParameters?: Record + signal?: AbortSignal +} + +export interface RunManagedAgentResult { + ok: boolean + content: string + sessionId?: string + error?: string + inputTokens?: number + outputTokens?: number +} + +type Terminal = { status: 'complete' | 'error'; reason?: string } +/** Result of handling one event: an optional terminal signal, plus whether the event must be retried. */ +type HandleResult = { terminal?: Terminal; retry?: boolean } + +/** True for session lifecycle events (`session.status_*` / `session.error`). */ +function isLifecycleEvent(type: string | undefined): boolean { + return !!type && (type.startsWith('session.status_') || type === 'session.error') +} + +/** + * Epoch millis for an event's `processed_at`. A missing/unparseable value + * counts as OLDEST (`-Infinity`) so an untimestamped lifecycle event can never + * outrank a timestamped one in either direction — a stray untimestamped event + * neither poisons the high-water mark (blocking later real events) nor clears a + * timestamped pause. Persisted lifecycle events always carry `processed_at`; + * this is purely defensive. + */ +function eventTime(value: string | null | undefined): number { + if (!value) return Number.NEGATIVE_INFINITY + const parsed = Date.parse(value) + return Number.isNaN(parsed) ? Number.NEGATIVE_INFINITY : parsed +} + +/** Best-effort `user.interrupt` for a session Sim is abandoning (cancel / cap). Never throws. */ +async function interruptQuietly(apiKey: string, sessionId: string): Promise { + try { + await interruptSession({ apiKey, sessionId }) + } catch (err) { + logger.warn('Failed to interrupt managed agent session on cancel', { + sessionId, + error: getErrorMessage(err), + }) + } +} + +export async function runManagedAgentSession( + input: RunManagedAgentInput +): Promise { + const { apiKey, signal } = input + if (signal?.aborted) return { ok: false, content: '', error: 'aborted' } + + const userMessage = input.userMessage.trim() + if (!userMessage) return { ok: false, content: '', error: 'User message is required.' } + + // Resolve the environment type up front so the payload routes correctly: + // self-hosted environments reject `resources`, so memory must go via metadata + // there. The authoritative source is the API; the block's hint is only a + // fallback if that lookup fails. + const environmentType = + (await getEnvironmentType({ apiKey, environmentId: input.environmentId, signal })) ?? + input.environmentType + + const createInput: CreateSessionInput = { + apiKey, + agentId: input.agentId, + environmentId: input.environmentId, + ...(environmentType ? { environmentType } : {}), + ...(input.title ? { title: input.title } : {}), + ...(input.vaultIds && input.vaultIds.length > 0 ? { vaultIds: input.vaultIds } : {}), + ...(input.memoryStoreId ? { memoryStoreId: input.memoryStoreId } : {}), + ...(input.memoryStoreId && input.memoryAccess ? { memoryAccess: input.memoryAccess } : {}), + ...(input.memoryStoreId && input.memoryInstructions + ? { memoryInstructions: input.memoryInstructions } + : {}), + ...(input.files && input.files.length > 0 ? { files: input.files } : {}), + ...(input.sessionParameters && Object.keys(input.sessionParameters).length > 0 + ? { sessionParameters: input.sessionParameters } + : {}), + ...(signal ? { signal } : {}), + } + + let sessionId: string + try { + const session = await createSession(createInput) + sessionId = session.id + } catch (error) { + return { + ok: false, + content: '', + error: getErrorMessage(error, 'Failed to create Managed Agent session'), + } + } + + logger.info('Created managed agent session', { + sessionId, + agentId: input.agentId, + environmentId: input.environmentId, + }) + + try { + await sendUserMessage({ apiKey, sessionId, text: userMessage, signal }) + } catch (error) { + // The session exists but we could not drive it. The message may still have + // reached the sandbox (abort/error can race the delivered request), so stop + // the session rather than leave it possibly running against the key. + await interruptQuietly(apiKey, sessionId) + return { + ok: false, + content: '', + sessionId, + error: signal?.aborted ? 'aborted' : getErrorMessage(error, 'Failed to send user message'), + } + } + + const assistantText = { value: '' } + const seenIds = new Set() + const startedAt = Date.now() + let backoffMs = RECONNECT_BACKOFF_START_MS + let sawActivity = false + // A `requires_action` idle means the session is paused waiting on a tool + // result, not finished. We derive it from the NEWEST lifecycle event by + // `processed_at` (tracked across the live stream and catch-up), so a lagging + // or out-of-order history can never clear a newer pause and let an idle + // snapshot report an in-progress session complete. + let latestLifecycleAt = Number.NEGATIVE_INFINITY + let requiresActionOutstanding = false + let terminal: Terminal | null = null + + const process = async (event: AnthropicSessionEvent): Promise => { + if ( + event.type === 'agent.message' || + event.type === 'agent.custom_tool_use' || + event.type === 'session.status_running' + ) { + sawActivity = true + } + if (isLifecycleEvent(event.type)) { + const at = eventTime(event.processed_at) + if (at >= latestLifecycleAt) { + latestLifecycleAt = at + requiresActionOutstanding = + event.type === 'session.status_idle' && event.stop_reason?.type === 'requires_action' + } + } + const outcome = await handleEvent({ event, assistantText, apiKey, sessionId, signal }) + // Mark the event seen only once fully handled. A custom-tool reply that + // failed to send stays unseen so the next catch-up retries it instead of + // stranding the session on an unanswered requires_action pause. + if (event.id && !outcome.retry) seenIds.add(event.id) + if (outcome.terminal) { + terminal = outcome.terminal + return true + } + return false + } + + try { + for (let iteration = 0; iteration < MAX_RECONNECT_ITERATIONS && !terminal; iteration++) { + if (signal?.aborted) break + if (Date.now() - startedAt > MAX_SESSION_MS) { + // Giving up on a session that may still be running — stop it so it + // does not keep consuming the workspace key past our cap. + await interruptQuietly(apiKey, sessionId) + terminal = { + status: 'error', + reason: `Session did not reach a terminal state within ${Math.floor(MAX_SESSION_MS / 1000)}s.`, + } + break + } + + const streamResp = await openSessionStream({ apiKey, sessionId, signal }) + await readSSEEvents(streamResp, { + signal, + onParseError: (raw, err) => { + logger.warn('Un-parseable SSE line', { + sessionId, + preview: raw.slice(0, 200), + error: getErrorMessage(err), + }) + }, + onEvent: async (event) => { + if (event.id && seenIds.has(event.id)) return undefined + return (await process(event)) ? true : undefined + }, + }) + if (terminal || signal?.aborted) break + + // Stream closed without a terminal event. Reconcile the full event + // history — the terminal event or final text may have landed during the + // gap. Events are deduped by id; entries without an id are skipped here + // (they are non-persisted stream previews, never history). + const history = await listSessionEvents({ apiKey, sessionId, signal }) + let progressed = false + for (const event of history) { + if (!event.id || seenIds.has(event.id)) continue + const isTerminal = await process(event) + // Count as progress only when the event was actually consumed (marked + // seen). A custom-tool reply that failed stays unseen for retry — it + // must NOT reset the backoff, or a persistently-failing reply would + // spin the reconnect loop with no delay. + if (seenIds.has(event.id)) progressed = true + if (isTerminal) break + } + if (terminal || signal?.aborted) break + + // Still no terminal event. Consult the authoritative session status: a + // finished session reports `idle`/`terminated`; a working one reports + // `running`. `idle` counts as complete only once the agent has actually + // started (a freshly-created session is `idle` before its first turn). + const snapshot = await getSession({ apiKey, sessionId, signal }) + if (snapshot?.status === 'terminated') { + terminal = { status: 'error', reason: 'Session terminated.' } + break + } + if (snapshot?.status === 'idle' && sawActivity && !requiresActionOutstanding) { + terminal = { status: 'complete' } + break + } + + // Working (or not yet started) — back off and reopen, resetting the + // backoff whenever catch-up surfaced new events. + if (progressed) { + backoffMs = RECONNECT_BACKOFF_START_MS + } else { + await sleep(backoffMs) + backoffMs = Math.min(backoffMs * 2, RECONNECT_BACKOFF_MAX_MS) + } + } + } catch (error) { + // Any exit from the loop with the session possibly still running must stop + // it — a mid-run stream/API failure is no different from an abort or cap. + await interruptQuietly(apiKey, sessionId) + if (signal?.aborted) { + return { ok: false, content: assistantText.value, sessionId, error: 'aborted' } + } + logger.error('Managed agent stream failed', { sessionId, error: getErrorMessage(error) }) + return { + ok: false, + content: assistantText.value, + sessionId, + error: getErrorMessage(error, 'Managed Agent session failed'), + } + } + + if (signal?.aborted) { + await interruptQuietly(apiKey, sessionId) + return { ok: false, content: assistantText.value, sessionId, error: 'aborted' } + } + if (!terminal) { + // Reconnect cap reached while the session may still be running — stop it so + // it does not keep consuming the workspace key after we give up. + await interruptQuietly(apiKey, sessionId) + return { + ok: false, + content: assistantText.value, + sessionId, + error: 'Reconnect iteration cap reached without a terminal state.', + } + } + if (terminal.status === 'error') { + return { + ok: false, + content: assistantText.value, + sessionId, + error: terminal.reason ?? 'Managed Agent session failed.', + } + } + + // Best-effort cumulative token usage for the block output. + const snapshot = await getSession({ apiKey, sessionId, signal }) + return { + ok: true, + content: assistantText.value, + sessionId, + ...(snapshot?.usage?.inputTokens !== undefined + ? { inputTokens: snapshot.usage.inputTokens } + : {}), + ...(snapshot?.usage?.outputTokens !== undefined + ? { outputTokens: snapshot.usage.outputTokens } + : {}), + } +} + +async function handleEvent(args: { + event: AnthropicSessionEvent + assistantText: { value: string } + apiKey: string + sessionId: string + signal?: AbortSignal +}): Promise { + const { event, assistantText, apiKey, sessionId, signal } = args + + if (event.type === 'agent.message') { + // Accumulate text only from persisted (id-bearing) messages. An idless + // `agent.message` is a stream-only preview that is never deduped, so + // appending it would double the text once the persisted copy arrives. + // Idless lifecycle/terminal events still flow through the handlers below. + if (event.id && Array.isArray(event.content)) { + for (const block of event.content) { + if (block?.type === 'text' && typeof block.text === 'string') { + assistantText.value += block.text + } + } + } + return {} + } + + if (event.type === 'agent.custom_tool_use') { + // Without an id we cannot correlate a reply; sending an empty id would + // strand the session, so log and let it resolve on its own instead. + if (!event.id) { + logger.warn('Managed Agent custom_tool_use arrived without an id — skipping reply', { + sessionId, + }) + return {} + } + logger.warn( + `Managed Agent invoked a custom tool "${event.name ?? ''}" that Sim does not provide — replying with error` + ) + try { + await sendSessionEvents({ + apiKey, + signal, + sessionId, + events: [ + { + type: 'user.custom_tool_result', + custom_tool_use_id: event.id, + content: [ + { + type: 'text', + text: 'This Managed Agent is being invoked from a Sim workflow block. Sim does not provide custom tools here — configure the agent to use only tools available in its Claude Platform workspace.', + }, + ], + is_error: true, + }, + ], + }) + } catch (err) { + logger.error('Failed to send custom_tool_result error reply', { + sessionId, + error: getErrorMessage(err), + }) + // Leave the event unseen so the next catch-up retries the reply rather + // than stranding the session on this unanswered tool call. + return { retry: true } + } + return {} + } + + if (event.type === 'session.status_terminated') { + return { terminal: { status: 'error', reason: event.error?.message ?? 'session_terminated' } } + } + + if (event.type === 'session.status_idle') { + const stop = event.stop_reason?.type + if (stop === 'end_turn') return { terminal: { status: 'complete' } } + if (stop === 'retries_exhausted') { + return { terminal: { status: 'error', reason: 'retries_exhausted' } } + } + // `requires_action` (paused for a pending tool call) and any unspecified + // idle are NOT terminal here. A freshly-created session is `idle` with no + // stop reason before its first turn, so completing on an unspecified idle + // would report empty content; instead defer to the authoritative-status + // gate, which only completes once `sawActivity` is set. + return {} + } + + if (event.type === 'session.error') { + return { + terminal: { + status: 'error', + reason: event.error?.message ?? event.message ?? 'session_error', + }, + } + } + + return {} +} diff --git a/apps/sim/lib/managed-agents/session-client.test.ts b/apps/sim/lib/managed-agents/session-client.test.ts new file mode 100644 index 00000000000..8eb870d3f69 --- /dev/null +++ b/apps/sim/lib/managed-agents/session-client.test.ts @@ -0,0 +1,191 @@ +/** + * @vitest-environment node + */ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { buildSessionCreatePayload, listSessionEvents } from '@/lib/managed-agents/session-client' + +const BASE = { + apiKey: 'sk-ant-fake', + agentId: 'agent_01ABC', + environmentId: 'env_01XYZ', +} as const + +describe('buildSessionCreatePayload — always-on fields', () => { + it('emits `agent` and `environment_id` from the input', () => { + expect(buildSessionCreatePayload({ ...BASE })).toEqual({ + agent: 'agent_01ABC', + environment_id: 'env_01XYZ', + }) + }) + + it('emits `title` when set', () => { + expect(buildSessionCreatePayload({ ...BASE, title: 'my session' }).title).toBe('my session') + }) + + it('emits `vault_ids` only when non-empty', () => { + expect(buildSessionCreatePayload({ ...BASE }).vault_ids).toBeUndefined() + expect(buildSessionCreatePayload({ ...BASE, vaultIds: [] }).vault_ids).toBeUndefined() + expect(buildSessionCreatePayload({ ...BASE, vaultIds: ['vlt_1', 'vlt_2'] }).vault_ids).toEqual([ + 'vlt_1', + 'vlt_2', + ]) + }) +}) + +describe('buildSessionCreatePayload — resources', () => { + it('attaches a memory store as a `memory_store` resource with default access', () => { + const payload = buildSessionCreatePayload({ ...BASE, memoryStoreId: 'memstore_01' }) + expect(payload.resources).toEqual([ + { type: 'memory_store', memory_store_id: 'memstore_01', access: 'read_write' }, + ]) + }) + + it('honors explicit read_only memory access', () => { + const payload = buildSessionCreatePayload({ + ...BASE, + memoryStoreId: 'memstore_01', + memoryAccess: 'read_only', + }) + expect(payload.resources).toEqual([ + { type: 'memory_store', memory_store_id: 'memstore_01', access: 'read_only' }, + ]) + }) + + it('includes memory instructions when provided', () => { + const payload = buildSessionCreatePayload({ + ...BASE, + memoryStoreId: 'memstore_01', + memoryInstructions: 'check before starting', + }) + expect(payload.resources).toEqual([ + { + type: 'memory_store', + memory_store_id: 'memstore_01', + access: 'read_write', + instructions: 'check before starting', + }, + ]) + }) + + it('attaches file resources with an optional mount path', () => { + const payload = buildSessionCreatePayload({ + ...BASE, + files: [{ fileId: 'file_1', mountPath: '/data/one' }, { fileId: 'file_2' }], + }) + expect(payload.resources).toEqual([ + { type: 'file', file_id: 'file_1', mount_path: '/data/one' }, + { type: 'file', file_id: 'file_2' }, + ]) + }) + + it('combines memory and file resources in order', () => { + const payload = buildSessionCreatePayload({ + ...BASE, + memoryStoreId: 'memstore_01', + files: [{ fileId: 'file_1' }], + }) + expect(payload.resources).toEqual([ + { type: 'memory_store', memory_store_id: 'memstore_01', access: 'read_write' }, + { type: 'file', file_id: 'file_1' }, + ]) + }) + + it('omits `resources` when nothing is attached', () => { + expect(buildSessionCreatePayload({ ...BASE }).resources).toBeUndefined() + expect(buildSessionCreatePayload({ ...BASE, files: [] }).resources).toBeUndefined() + }) +}) + +describe('buildSessionCreatePayload — metadata', () => { + it('emits `metadata` from sessionParameters', () => { + const payload = buildSessionCreatePayload({ + ...BASE, + sessionParameters: { foo: 'bar', baz: 'qux' }, + }) + expect(payload.metadata).toEqual({ foo: 'bar', baz: 'qux' }) + }) + + it('omits `metadata` when there are no session parameters', () => { + expect(buildSessionCreatePayload({ ...BASE }).metadata).toBeUndefined() + expect(buildSessionCreatePayload({ ...BASE, sessionParameters: {} }).metadata).toBeUndefined() + }) + + it('keeps memory on resources and never folds it into metadata (cloud)', () => { + const payload = buildSessionCreatePayload({ + ...BASE, + memoryStoreId: 'memstore_01', + sessionParameters: { env: 'staging' }, + }) + expect(payload.metadata).toEqual({ env: 'staging' }) + expect(payload.resources).toEqual([ + { type: 'memory_store', memory_store_id: 'memstore_01', access: 'read_write' }, + ]) + }) +}) + +describe('buildSessionCreatePayload — self-hosted routing', () => { + it('never sends resources on self-hosted and does not auto-route memory (no native support)', () => { + const payload = buildSessionCreatePayload({ + ...BASE, + environmentType: 'self_hosted', + memoryStoreId: 'memstore_01', + memoryAccess: 'read_only', + memoryInstructions: 'use it', + files: [{ fileId: 'file_1' }], + sessionParameters: { SOURCE_TYPE: 'git' }, + }) + expect(payload.resources).toBeUndefined() + // Only the author's explicit metadata is forwarded — memory is NOT injected. + expect(payload.metadata).toEqual({ SOURCE_TYPE: 'git' }) + }) + + it('sends no metadata on self-hosted when the author set none', () => { + const payload = buildSessionCreatePayload({ + ...BASE, + environmentType: 'self_hosted', + memoryStoreId: 'memstore_01', + files: [{ fileId: 'file_1' }], + }) + expect(payload.resources).toBeUndefined() + expect(payload.metadata).toBeUndefined() + }) + + it('cloud (default) still attaches memory + files as resources', () => { + const payload = buildSessionCreatePayload({ + ...BASE, + environmentType: 'cloud', + memoryStoreId: 'memstore_01', + files: [{ fileId: 'file_1' }], + }) + expect(payload.resources).toEqual([ + { type: 'memory_store', memory_store_id: 'memstore_01', access: 'read_write' }, + { type: 'file', file_id: 'file_1' }, + ]) + expect(payload.metadata).toBeUndefined() + }) +}) + +describe('listSessionEvents — ordering', () => { + const originalFetch = global.fetch + afterEach(() => { + global.fetch = originalFetch + }) + + it('orders events by processed_at ascending, with queued (null) events last', async () => { + global.fetch = vi.fn(async () => + Response.json({ + data: [ + { id: 'c', type: 'agent.message', processed_at: '2026-01-01T00:00:03Z' }, + { id: 'a', type: 'agent.message', processed_at: '2026-01-01T00:00:01Z' }, + { id: 'queued', type: 'agent.message', processed_at: null }, + { id: 'b', type: 'agent.message', processed_at: '2026-01-01T00:00:02Z' }, + ], + next_page: null, + }) + ) as unknown as typeof fetch + + const events = await listSessionEvents({ apiKey: 'sk-ant-fake', sessionId: 'sess_1' }) + + expect(events.map((e) => e.id)).toEqual(['a', 'b', 'c', 'queued']) + }) +}) diff --git a/apps/sim/lib/managed-agents/session-client.ts b/apps/sim/lib/managed-agents/session-client.ts new file mode 100644 index 00000000000..e03a6d3827e --- /dev/null +++ b/apps/sim/lib/managed-agents/session-client.ts @@ -0,0 +1,425 @@ +/** + * Provider-neutral HTTP client for the Claude Platform Managed Agents API. + * + * A thin wrapper around `fetch` that speaks the Managed Agents beta. It has + * NO Sim-domain dependencies (no `@sim/db`, no encryption, no executor + * types) so it can be unit-tested in isolation and imported from either the + * server run route or the block-editor proxy route. + * + * Shapes are validated against the Claude Platform docs: + * https://platform.claude.com/docs/en/managed-agents/ + */ + +export const ANTHROPIC_API_BASE = 'https://api.anthropic.com' +export const ANTHROPIC_VERSION = '2023-06-01' +/** Beta header for every session/agent/environment/vault endpoint. */ +export const MANAGED_AGENTS_BETA = 'managed-agents-2026-04-01' +/** + * Memory-store endpoints require a DIFFERENT beta header, and combining it + * with {@link MANAGED_AGENTS_BETA} on the same request is a documented 400. + * https://platform.claude.com/docs/en/managed-agents/memory + */ +export const AGENT_MEMORY_BETA = 'agent-memory-2026-07-22' + +/** + * Minimal shape of a session event as delivered over SSE or the events list. + * The run loop only reads these fields, so we model them structurally rather + * than exhaustively. + */ +export interface AnthropicSessionEvent { + id?: string + type?: string + content?: Array<{ type: string; text?: string }> + name?: string + stop_reason?: { type?: string } + error?: { message?: string } + message?: string + /** Server-side record time; `null`/absent means still queued (handled after processed events). */ + processed_at?: string | null +} + +/** + * Shared inputs on every managed-agents call. `apiKey` is the caller's + * Claude Platform API key (an Anthropic workspace-scoped key); `signal` + * propagates cancellation into the outbound fetch. + */ +export interface SessionAuth { + apiKey: string + signal?: AbortSignal +} + +export interface CreateSessionInput extends SessionAuth { + agentId: string + environmentId: string + /** + * Environment execution model. Self-hosted environments reject the + * `resources` array, so memory is routed via `metadata` and files are + * dropped for them. Defaults to cloud behavior when unset. + */ + environmentType?: EnvironmentType + /** Optional session title stored on the Anthropic session. */ + title?: string + /** OAuth credential vaults the agent's MCP tools can reference. */ + vaultIds?: string[] + /** Memory-store id (`memstore_...`) attached as a session resource. */ + memoryStoreId?: string + /** Access mode on the attached memory store. Ignored when `memoryStoreId` is unset. */ + memoryAccess?: 'read_write' | 'read_only' + /** Per-attachment guidance rendered into the memory section of the system prompt. */ + memoryInstructions?: string + /** Files-API files (`file_...`) attached as `file` session resources. */ + files?: Array<{ fileId: string; mountPath?: string }> + /** Arbitrary session metadata (wire name: `metadata`). */ + sessionParameters?: Record +} + +export interface CreateSessionResult { + id: string +} + +/** Cumulative token usage returned on the session resource. */ +export interface SessionUsage { + inputTokens?: number + outputTokens?: number +} + +/** Environment execution model per `GET /v1/environments/{id}` → `config.type`. */ +export type EnvironmentType = 'cloud' | 'self_hosted' + +/** Authoritative session status per `GET /v1/sessions/{id}`. */ +export type SessionStatus = 'idle' | 'running' | 'rescheduling' | 'terminated' + +export interface SessionSnapshot { + status?: SessionStatus + usage?: SessionUsage +} + +/** + * Standard header set for Managed Agents calls. `beta` overrides the default + * managed-agents beta for memory-store endpoints. Only ONE beta value is ever + * sent — combining the two is a documented 400. + */ +function managedAgentsHeaders( + apiKey: string, + options: { json?: boolean; accept?: string; beta?: string } = {} +): Record { + const headers: Record = { + 'x-api-key': apiKey, + 'anthropic-version': ANTHROPIC_VERSION, + 'anthropic-beta': options.beta ?? MANAGED_AGENTS_BETA, + } + if (options.json) headers['content-type'] = 'application/json' + if (options.accept) headers.accept = options.accept + return headers +} + +/** + * Builds the request body for `POST /v1/sessions`. + * + * Cloud environments attach memory stores and files via the `resources[]` + * array. Self-hosted environments REJECT `resources` (a documented 400 — + * "resources are not supported with self-hosted environments") and have no + * native memory/file attach, so those are omitted there; the block hides the + * fields accordingly. Session parameters always go on `metadata` for both — a + * self-hosted worker that consumes a memory store reads it from a metadata key + * the author sets explicitly. + */ +export function buildSessionCreatePayload(input: CreateSessionInput): Record { + const payload: Record = { + agent: input.agentId, + environment_id: input.environmentId, + } + if (input.title) payload.title = input.title + if (input.vaultIds && input.vaultIds.length > 0) payload.vault_ids = input.vaultIds + + // `resources` (memory stores + files) are cloud-only. Self-hosted rejects them. + if (input.environmentType !== 'self_hosted') { + const resources: Array> = [] + if (input.memoryStoreId) { + const memory: Record = { + type: 'memory_store', + memory_store_id: input.memoryStoreId, + access: input.memoryAccess ?? 'read_write', + } + if (input.memoryInstructions) memory.instructions = input.memoryInstructions + resources.push(memory) + } + if (input.files && input.files.length > 0) { + for (const file of input.files) { + if (!file.fileId) continue + const entry: Record = { type: 'file', file_id: file.fileId } + if (file.mountPath) entry.mount_path = file.mountPath + resources.push(entry) + } + } + if (resources.length > 0) payload.resources = resources + } + + if (input.sessionParameters && Object.keys(input.sessionParameters).length > 0) { + payload.metadata = { ...input.sessionParameters } + } + return payload +} + +/** + * POST /v1/sessions — provisions a session sandbox. Does NOT start work; a + * subsequent `sendUserMessage` is what causes the agent to run. + */ +export async function createSession(input: CreateSessionInput): Promise { + const resp = await fetch(`${ANTHROPIC_API_BASE}/v1/sessions`, { + method: 'POST', + headers: managedAgentsHeaders(input.apiKey, { json: true }), + body: JSON.stringify(buildSessionCreatePayload(input)), + signal: input.signal, + }) + if (!resp.ok) { + const detail = await resp.text().catch(() => '') + throw new Error(`Anthropic sessions.create failed (${resp.status}): ${detail.slice(0, 400)}`) + } + const body = (await resp.json()) as { id?: unknown } + if (typeof body.id !== 'string' || body.id.length === 0) { + throw new Error('Anthropic sessions.create returned no id') + } + return { id: body.id } +} + +interface UserMessageEvent { + type: 'user.message' + content: Array<{ type: 'text'; text: string }> +} + +interface UserCustomToolResultEvent { + type: 'user.custom_tool_result' + custom_tool_use_id: string + content: Array<{ type: 'text'; text: string }> + is_error: boolean +} + +/** Stops a running session mid-execution; the session stays usable afterward. */ +interface UserInterruptEvent { + type: 'user.interrupt' +} + +export type OutboundSessionEvent = UserMessageEvent | UserCustomToolResultEvent | UserInterruptEvent + +/** POST /v1/sessions/{id}/events with a single `user.message`. */ +export async function sendUserMessage( + input: SessionAuth & { sessionId: string; text: string } +): Promise { + await sendSessionEvents({ + apiKey: input.apiKey, + signal: input.signal, + sessionId: input.sessionId, + events: [{ type: 'user.message', content: [{ type: 'text', text: input.text }] }], + }) +} + +/** Generic events-send used for both `user.message` and `user.custom_tool_result`. */ +export async function sendSessionEvents( + input: SessionAuth & { sessionId: string; events: OutboundSessionEvent[] } +): Promise { + if (input.events.length === 0) return + const resp = await fetch(`${ANTHROPIC_API_BASE}/v1/sessions/${input.sessionId}/events`, { + method: 'POST', + headers: managedAgentsHeaders(input.apiKey, { json: true }), + body: JSON.stringify({ events: input.events }), + signal: input.signal, + }) + if (!resp.ok) { + const detail = await resp.text().catch(() => '') + throw new Error(`Anthropic events.send failed (${resp.status}): ${detail.slice(0, 400)}`) + } +} + +/** Best-effort timeout for the fire-on-cancel interrupt (its own, since the run signal is already aborted). */ +const INTERRUPT_TIMEOUT_MS = 5000 + +/** + * POST /v1/sessions/{id}/events with a `user.interrupt` — stops a session that + * is still running so it stops consuming the workspace API key once Sim has + * given up on it (workflow cancelled or wall-clock cap hit). Deliberately uses + * its OWN short timeout rather than the run's abort signal, which is already + * aborted by the time this fires. + */ +export async function interruptSession(input: { + apiKey: string + sessionId: string +}): Promise { + await sendSessionEvents({ + apiKey: input.apiKey, + sessionId: input.sessionId, + events: [{ type: 'user.interrupt' }], + signal: AbortSignal.timeout(INTERRUPT_TIMEOUT_MS), + }) +} + +/** GET /v1/sessions/{id}/events/stream — opens the SSE response. */ +export async function openSessionStream( + input: SessionAuth & { sessionId: string } +): Promise { + const resp = await fetch(`${ANTHROPIC_API_BASE}/v1/sessions/${input.sessionId}/events/stream`, { + method: 'GET', + headers: managedAgentsHeaders(input.apiKey, { accept: 'text/event-stream' }), + signal: input.signal, + }) + if (!resp.ok) { + const detail = await resp.text().catch(() => '') + throw new Error(`Anthropic events.stream failed (${resp.status}): ${detail.slice(0, 400)}`) + } + if (!resp.body) throw new Error('Anthropic events.stream returned no body') + return resp +} + +/** A single page of a Managed Agents list endpoint (page-cursor pagination). */ +interface AnthropicListPage { + data?: T[] + next_page?: string | null +} + +/** + * Drains a page-cursor-paginated list endpoint (`?limit=&page=` following + * `next_page` until null). Used for both the block-editor dropdowns and the + * session-event catch-up. `beta` overrides the default header for memory + * stores. + */ +const MAX_LIST_PAGES = 1000 + +async function listPaginated( + input: SessionAuth & { path: string; beta?: string; maxItems?: number } +): Promise { + const collected: T[] = [] + const maxItems = input.maxItems ?? 2000 + let page: string | null = null + // `MAX_LIST_PAGES` bounds a misbehaving cursor that never returns `next_page: + // null`; real histories terminate well before it. + for (let pageCount = 0; pageCount < MAX_LIST_PAGES && collected.length < maxItems; pageCount++) { + const url = new URL(`${ANTHROPIC_API_BASE}${input.path}`) + url.searchParams.set('limit', '100') + if (page) url.searchParams.set('page', page) + const resp = await fetch(url.toString(), { + method: 'GET', + headers: managedAgentsHeaders(input.apiKey, { beta: input.beta }), + signal: input.signal, + }) + if (!resp.ok) { + const detail = await resp.text().catch(() => '') + throw new Error(`Anthropic ${input.path} failed (${resp.status}): ${detail.slice(0, 400)}`) + } + const body = (await resp.json()) as AnthropicListPage + const items = Array.isArray(body.data) ? body.data : [] + collected.push(...items) + if (!body.next_page || items.length === 0) break + page = body.next_page + } + return collected +} + +/** + * Full event history for a session (`GET /v1/sessions/{id}/events`), used by + * the reconnect/catch-up loop to recover events missed while the SSE stream + * was closed. The caller dedups against already-seen event ids. Drains every + * page so the tail (terminal status / final assistant text) is never cut off + * by a page cap. + */ +export async function listSessionEvents( + input: SessionAuth & { sessionId: string } +): Promise { + const events = await listPaginated({ + apiKey: input.apiKey, + signal: input.signal, + path: `/v1/sessions/${input.sessionId}/events`, + maxItems: Number.POSITIVE_INFINITY, + }) + // The list endpoint's page order is not guaranteed chronological, so order by + // the server-side `processed_at` timestamp before returning. The catch-up + // loop depends on ascending order both to accumulate assistant text in order + // and to read the latest lifecycle event. Still-queued events (null + // `processed_at`) are processed after everything else, so they sort last. + return events.sort((a, b) => parseProcessedAt(a.processed_at) - parseProcessedAt(b.processed_at)) +} + +/** Epoch millis for a `processed_at`, or +Infinity when absent/queued/unparseable (sorts last). */ +function parseProcessedAt(value: string | null | undefined): number { + if (!value) return Number.POSITIVE_INFINITY + const parsed = Date.parse(value) + return Number.isNaN(parsed) ? Number.POSITIVE_INFINITY : parsed +} + +/** + * Lists a Managed Agents resource collection for the block-editor dropdowns. + * Memory stores require the agent-memory beta header; everything else uses the + * managed-agents beta. + */ +export async function managedAgentsList( + input: SessionAuth & { path: string; beta?: string } +): Promise { + return listPaginated({ + apiKey: input.apiKey, + signal: input.signal, + path: input.path, + beta: input.beta, + }) +} + +/** + * GET /v1/environments/{id} — resolves the environment's execution model from + * `config.type`. Drives session-payload routing: self-hosted rejects + * `resources`. Returns `undefined` on any error so the caller can fall back to + * cloud behavior. + */ +export async function getEnvironmentType( + input: SessionAuth & { environmentId: string } +): Promise { + try { + const resp = await fetch(`${ANTHROPIC_API_BASE}/v1/environments/${input.environmentId}`, { + method: 'GET', + headers: managedAgentsHeaders(input.apiKey), + signal: input.signal, + }) + if (!resp.ok) return undefined + const body = (await resp.json()) as { config?: { type?: unknown } } + const type = body.config?.type + return type === 'cloud' || type === 'self_hosted' ? type : undefined + } catch { + return undefined + } +} + +/** + * GET /v1/sessions/{id} — retrieves the session resource. Returns the + * authoritative `status` (used to decide completion when the event stream is + * quiet) and cumulative token `usage` (surfaced as block output). Returns + * `null` on any error so callers can treat it as best-effort. + */ +export async function getSession( + input: SessionAuth & { sessionId: string } +): Promise { + try { + const resp = await fetch(`${ANTHROPIC_API_BASE}/v1/sessions/${input.sessionId}`, { + method: 'GET', + headers: managedAgentsHeaders(input.apiKey), + signal: input.signal, + }) + if (!resp.ok) return null + const body = (await resp.json()) as { + status?: unknown + usage?: { input_tokens?: unknown; output_tokens?: unknown } + } + const snapshot: SessionSnapshot = {} + if ( + body.status === 'idle' || + body.status === 'running' || + body.status === 'rescheduling' || + body.status === 'terminated' + ) { + snapshot.status = body.status + } + const usage: SessionUsage = {} + if (typeof body.usage?.input_tokens === 'number') usage.inputTokens = body.usage.input_tokens + if (typeof body.usage?.output_tokens === 'number') usage.outputTokens = body.usage.output_tokens + if (usage.inputTokens !== undefined || usage.outputTokens !== undefined) snapshot.usage = usage + return snapshot + } catch { + return null + } +} diff --git a/apps/sim/lib/managed-agents/subblock-options.ts b/apps/sim/lib/managed-agents/subblock-options.ts new file mode 100644 index 00000000000..ca25119ef56 --- /dev/null +++ b/apps/sim/lib/managed-agents/subblock-options.ts @@ -0,0 +1,64 @@ +import { requestJson } from '@/lib/api/client/request' +import { + listManagedAgentOptionsContract, + type ManagedAgentOption, + type ManagedAgentResource, +} from '@/lib/api/contracts/managed-agents' +import { useWorkflowRegistry } from '@/stores/workflows/registry/store' +import { useSubBlockStore } from '@/stores/workflows/subblock/store' + +/** + * `fetchOptions` helpers for the Managed Agent block's dropdowns. Each reads + * the block's selected Claude Platform `credential` and calls the list route, + * which decrypts the credential's key server-side — the API key never touches + * the browser. + */ + +function readSubBlockValue(blockId: string, key: string): string | null { + const activeWorkflowId = useWorkflowRegistry.getState().activeWorkflowId + if (!activeWorkflowId) return null + const value = useSubBlockStore.getState().workflowValues[activeWorkflowId]?.[blockId]?.[key] + return typeof value === 'string' && value.length > 0 ? value : null +} + +async function fetchOptions( + blockId: string, + resource: ManagedAgentResource +): Promise { + const credentialId = readSubBlockValue(blockId, 'credential') + if (!credentialId) return [] + try { + const { options } = await requestJson(listManagedAgentOptionsContract, { + query: { credentialId, resource }, + }) + return options + } catch { + return [] + } +} + +export function fetchManagedAgentAgentOptions(blockId: string): Promise { + return fetchOptions(blockId, 'agents') +} + +export async function fetchManagedAgentEnvironmentOptions( + blockId: string +): Promise { + const options = await fetchOptions(blockId, 'environments') + // Filter to the selected environment type so cloud/self-hosted stay separate + // (self-hosted rejects `resources`, so the two modes expose different fields). + // Options with an unknown type are kept as a safety net. + const mode = readSubBlockValue(blockId, 'environmentType') + if (mode !== 'cloud' && mode !== 'self_hosted') return options + return options.filter((option) => option.type === undefined || option.type === mode) +} + +export function fetchManagedAgentVaultOptions(blockId: string): Promise { + return fetchOptions(blockId, 'vaults') +} + +export function fetchManagedAgentMemoryStoreOptions( + blockId: string +): Promise { + return fetchOptions(blockId, 'memory-stores') +} diff --git a/apps/sim/lib/oauth/oauth.ts b/apps/sim/lib/oauth/oauth.ts index 60b6815cd3b..99477a53525 100644 --- a/apps/sim/lib/oauth/oauth.ts +++ b/apps/sim/lib/oauth/oauth.ts @@ -8,6 +8,7 @@ import { AzureIcon, BoxCompanyIcon, CalComIcon, + ClaudeIcon, ClickUpIcon, ConfluenceIcon, DocuSignIcon, @@ -66,6 +67,23 @@ import type { OAuthProviderConfig } from './types' const logger = createLogger('OAuth') export const OAUTH_PROVIDERS: Record = { + 'claude-platform': { + name: 'Claude Platform', + icon: ClaudeIcon, + services: { + 'claude-platform': { + name: 'Claude Platform', + description: 'Run Claude Platform Managed Agents from your workflows.', + providerId: 'claude-platform', + serviceAccountProviderId: 'claude-platform-service-account', + icon: ClaudeIcon, + baseProviderIcon: ClaudeIcon, + scopes: [], + authType: 'service_account', + }, + }, + defaultService: 'claude-platform', + }, google: { name: 'Google', icon: GoogleIcon, diff --git a/apps/sim/lib/oauth/utils.ts b/apps/sim/lib/oauth/utils.ts index befbc8b5065..641355949b0 100644 --- a/apps/sim/lib/oauth/utils.ts +++ b/apps/sim/lib/oauth/utils.ts @@ -609,6 +609,18 @@ for (const [baseProviderId, providerConfig] of Object.entries(OAUTH_PROVIDERS)) baseProvider: baseProviderId, serviceKey, } + // Service-account credentials are stored under `serviceAccountProviderId` + // (e.g. `claude-platform-service-account`). Map it to the same base so + // icon/name resolution doesn't fall back to a mis-split base provider — the + // hyphen split only recovers a single-segment base (`google`), not a + // multi-segment one (`claude-platform`). First service to claim it wins. + const saProviderId = service.serviceAccountProviderId + if (saProviderId && !PROVIDER_ID_TO_BASE_PROVIDER[saProviderId]) { + PROVIDER_ID_TO_BASE_PROVIDER[saProviderId] = { + baseProvider: baseProviderId, + serviceKey, + } + } } } diff --git a/apps/sim/tools/index.ts b/apps/sim/tools/index.ts index adb32d63f5e..165e8ed4d26 100644 --- a/apps/sim/tools/index.ts +++ b/apps/sim/tools/index.ts @@ -1240,7 +1240,7 @@ export async function executeTool( // Check for direct execution (no HTTP request needed) if (tool.directExecution) { logger.info(`[${requestId}] Using directExecution for ${toolId}`) - const result = await tool.directExecution(contextParams) + const result = await tool.directExecution(contextParams, effectiveSignal) // Apply post-processing if available and not skipped let finalResult = result diff --git a/apps/sim/tools/managed_agent/index.ts b/apps/sim/tools/managed_agent/index.ts new file mode 100644 index 00000000000..2b2a3212ada --- /dev/null +++ b/apps/sim/tools/managed_agent/index.ts @@ -0,0 +1,2 @@ +export { managedAgentRunSessionTool } from './run_session' +export * from './types' diff --git a/apps/sim/tools/managed_agent/normalizers.test.ts b/apps/sim/tools/managed_agent/normalizers.test.ts new file mode 100644 index 00000000000..8024fc42806 --- /dev/null +++ b/apps/sim/tools/managed_agent/normalizers.test.ts @@ -0,0 +1,102 @@ +/** + * @vitest-environment node + */ +import { describe, expect, it } from 'vitest' +import { + isTruthyAck, + normalizeFiles, + normalizeMemoryAccess, + normalizeSessionParameters, + normalizeStringList, +} from '@/tools/managed_agent/normalizers' + +describe('isTruthyAck', () => { + it('accepts boolean true and common string forms', () => { + for (const v of [true, 'true', 'TRUE', ' 1 ', 'yes']) expect(isTruthyAck(v)).toBe(true) + }) + it('rejects everything else', () => { + for (const v of [false, 'false', '0', '', undefined, null, 1, {}]) { + expect(isTruthyAck(v)).toBe(false) + } + }) +}) + +describe('normalizeMemoryAccess', () => { + it('passes through valid modes and drops others', () => { + expect(normalizeMemoryAccess('read_write')).toBe('read_write') + expect(normalizeMemoryAccess('read_only')).toBe('read_only') + expect(normalizeMemoryAccess('nonsense')).toBeUndefined() + expect(normalizeMemoryAccess(undefined)).toBeUndefined() + }) +}) + +describe('normalizeStringList', () => { + it('handles arrays, json strings, comma-lists, and single strings', () => { + expect(normalizeStringList(['a', ' b ', ''])).toEqual(['a', 'b']) + expect(normalizeStringList('["x","y"]')).toEqual(['x', 'y']) + expect(normalizeStringList('a, b ,c')).toEqual(['a', 'b', 'c']) + expect(normalizeStringList('solo')).toEqual(['solo']) + expect(normalizeStringList('')).toEqual([]) + expect(normalizeStringList(undefined)).toEqual([]) + }) +}) + +describe('normalizeFiles', () => { + it('reads the File ID / Mount path column shape', () => { + const rows = [ + { cells: { 'File ID': 'file_1', 'Mount path': '/a' } }, + { cells: { 'File ID': ' file_2 ' } }, + ] + expect(normalizeFiles(rows)).toEqual([ + { fileId: 'file_1', mountPath: '/a' }, + { fileId: 'file_2' }, + ]) + }) + it('accepts the flat shape and drops rows without a file id', () => { + expect(normalizeFiles([{ fileId: 'file_ok' }, { fileId: '' }, {}])).toEqual([ + { fileId: 'file_ok' }, + ]) + }) + it('accepts plain string arrays and comma lists', () => { + expect(normalizeFiles(['file_1', ' file_2 '])).toEqual([ + { fileId: 'file_1' }, + { fileId: 'file_2' }, + ]) + expect(normalizeFiles('file_1, file_2')).toEqual([{ fileId: 'file_1' }, { fileId: 'file_2' }]) + }) + it('parses a JSON-stringified table of rows instead of dropping it', () => { + expect(normalizeFiles('[{"cells":{"File ID":"file_1","Mount path":"/a"}}]')).toEqual([ + { fileId: 'file_1', mountPath: '/a' }, + ]) + expect(normalizeFiles('[{"fileId":"file_2"}]')).toEqual([{ fileId: 'file_2' }]) + // A JSON array of plain strings still works. + expect(normalizeFiles('["file_3"]')).toEqual([{ fileId: 'file_3' }]) + }) + it('returns [] for empty input', () => { + expect(normalizeFiles(undefined)).toEqual([]) + expect(normalizeFiles('')).toEqual([]) + }) +}) + +describe('normalizeSessionParameters', () => { + it('reads table rows keyed by Key/Value', () => { + const rows = [{ cells: { Key: 'A', Value: '1' } }, { cells: { Key: 'B', Value: '2' } }] + expect(normalizeSessionParameters(rows)).toEqual({ A: '1', B: '2' }) + }) + it('accepts a flat object and a json string', () => { + expect(normalizeSessionParameters({ A: '1' })).toEqual({ A: '1' }) + expect(normalizeSessionParameters('[{"cells":{"Key":"A","Value":"1"}}]')).toEqual({ A: '1' }) + }) + it('stringifies scalar values from a flat object and drops non-scalars', () => { + expect(normalizeSessionParameters({ N: 42, B: true, O: { x: 1 } })).toEqual({ + N: '42', + B: 'true', + O: '', + }) + }) + it('drops blank keys and returns undefined when empty', () => { + expect(normalizeSessionParameters([{ cells: { Key: ' ', Value: 'x' } }])).toBeUndefined() + expect(normalizeSessionParameters([])).toBeUndefined() + expect(normalizeSessionParameters(undefined)).toBeUndefined() + }) +}) diff --git a/apps/sim/tools/managed_agent/normalizers.ts b/apps/sim/tools/managed_agent/normalizers.ts new file mode 100644 index 00000000000..0b45be961b6 --- /dev/null +++ b/apps/sim/tools/managed_agent/normalizers.ts @@ -0,0 +1,167 @@ +/** + * Pure normalization helpers that shape the Managed Agent block's subblock + * values (which arrive in several runtime shapes — table rows, JSON strings, + * flat objects, comma-lists) into the tidy typed values the session runner + * expects. No server deps so each helper is directly unit-testable. + */ + +export function normalizeMemoryAccess(value: unknown): 'read_write' | 'read_only' | undefined { + if (value === 'read_write' || value === 'read_only') return value + return undefined +} + +/** + * A `switch` value may arrive as a real boolean or as a string (`"true"`, + * `"1"`, `"yes"`) depending on serialization. Treat every reasonable + * "checked" form as truthy; anything else as not-checked. + */ +export function isTruthyAck(value: unknown): boolean { + if (value === true) return true + if (typeof value !== 'string') return false + const normalized = value.trim().toLowerCase() + return normalized === 'true' || normalized === '1' || normalized === 'yes' +} + +/** + * Coerces the block's file table into `{ fileId, mountPath? }[]` for the + * session `resources` array (`{ type: 'file', file_id, mount_path? }`). Reads + * the `File ID` / `Mount path` column shape, the flat `{ fileId, mountPath }` + * shape, and plain string / comma / json id lists. Drops blank ids. + */ +export function normalizeFiles(value: unknown): Array<{ fileId: string; mountPath?: string }> { + // A table subblock can arrive JSON-stringified. Parse a leading-'[' string + // into its array form first, so rows of objects aren't mistaken for a plain + // string list (which would silently drop every file attachment). + let normalized: unknown = value + if (typeof normalized === 'string' && normalized.trim().startsWith('[')) { + try { + normalized = JSON.parse(normalized.trim()) + } catch { + // Not valid JSON — leave as a string; handled as a comma/single id below. + } + } + if ( + typeof normalized === 'string' || + (Array.isArray(normalized) && normalized.every((v) => typeof v === 'string')) + ) { + return normalizeStringList(normalized).map((fileId) => ({ fileId })) + } + if (!Array.isArray(normalized)) return [] + const out: Array<{ fileId: string; mountPath?: string }> = [] + for (const raw of normalized) { + if (typeof raw === 'string') { + if (raw.trim()) out.push({ fileId: raw.trim() }) + continue + } + if (!raw || typeof raw !== 'object') continue + const record = raw as Record + const cells = + record.cells && typeof record.cells === 'object' + ? (record.cells as Record) + : record + const readString = (key: string): string | undefined => + typeof cells[key] === 'string' ? (cells[key] as string) : undefined + const fileId = readString('fileId') ?? readString('File ID') ?? readString('file_id') ?? '' + if (!fileId.trim()) continue + const mountPath = + readString('mountPath') ?? readString('Mount path') ?? readString('mount_path') + out.push({ + fileId: fileId.trim(), + ...(mountPath?.trim() ? { mountPath: mountPath.trim() } : {}), + }) + } + return out +} + +/** + * Coerce a multi-select combobox / json input — array, JSON-encoded array + * string, comma-separated string, or single string — into a trimmed + * `string[]`. + */ +export function normalizeStringList(value: unknown): string[] { + if (Array.isArray(value)) { + return value + .filter((v): v is string => typeof v === 'string' && v.trim().length > 0) + .map((v) => v.trim()) + } + if (typeof value !== 'string') return [] + const trimmed = value.trim() + if (!trimmed) return [] + if (trimmed.startsWith('[')) { + // Meant to be a JSON array — parse it, but do NOT comma-split the raw JSON + // text on failure (that yields garbage tokens like `["x"`). An empty result + // is the honest outcome for a malformed/non-array JSON string. + try { + const parsed = JSON.parse(trimmed) + if (Array.isArray(parsed)) { + return parsed + .filter((v): v is string => typeof v === 'string' && v.trim().length > 0) + .map((v) => v.trim()) + } + } catch { + // fall through to [] + } + return [] + } + return trimmed + .split(',') + .map((v) => v.trim()) + .filter((v) => v.length > 0) +} + +/** + * Coerce the block's metadata table into `Record` for the + * session `metadata` field. Accepts `WorkflowTableRow[]`, a JSON-encoded + * array string, or a flat object. Drops rows with a blank key. + */ +export function normalizeSessionParameters(value: unknown): Record | undefined { + const rows = coerceToRows(value) + if (rows === undefined) return undefined + const out: Record = {} + for (const row of rows) { + const key = typeof row.key === 'string' ? row.key.trim() : '' + if (!key) continue + // Metadata is a string map. Preserve scalar values (a flat object can carry + // numbers/booleans) by stringifying them; drop non-scalars to empty. + const raw = row.value + out[key] = + typeof raw === 'string' + ? raw + : typeof raw === 'number' || typeof raw === 'boolean' + ? String(raw) + : '' + } + return Object.keys(out).length > 0 ? out : undefined +} + +function coerceToRows(value: unknown): Array<{ key: unknown; value: unknown }> | undefined { + if (Array.isArray(value)) return value.map((row) => tableRowToPair(row)) + if (typeof value === 'string') { + const trimmed = value.trim() + if (!trimmed || !trimmed.startsWith('[')) return undefined + try { + const parsed = JSON.parse(trimmed) + if (Array.isArray(parsed)) return parsed.map((row) => tableRowToPair(row)) + } catch { + return undefined + } + return undefined + } + if (value && typeof value === 'object') { + return Object.entries(value as Record).map(([key, val]) => ({ + key, + value: val, + })) + } + return undefined +} + +function tableRowToPair(row: unknown): { key: unknown; value: unknown } { + if (!row || typeof row !== 'object') return { key: undefined, value: undefined } + const record = row as Record + const cells = + record.cells && typeof record.cells === 'object' + ? (record.cells as Record) + : record + return { key: cells.Key ?? cells.key, value: cells.Value ?? cells.value } +} diff --git a/apps/sim/tools/managed_agent/run_session.test.ts b/apps/sim/tools/managed_agent/run_session.test.ts new file mode 100644 index 00000000000..9abbb72ceb3 --- /dev/null +++ b/apps/sim/tools/managed_agent/run_session.test.ts @@ -0,0 +1,78 @@ +/** + * @vitest-environment node + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { runManagedAgentSession } = vi.hoisted(() => ({ + runManagedAgentSession: vi.fn(), +})) + +vi.mock('@/lib/managed-agents/run-session', () => ({ runManagedAgentSession })) + +import { managedAgentRunSessionTool } from '@/tools/managed_agent/run_session' +import type { ManagedAgentRunSessionParams } from '@/tools/managed_agent/types' + +const run = (params: Partial) => + managedAgentRunSessionTool.directExecution!({ + credential: 'cred_1', + accessToken: 'sk-ant-fake', + agent: 'agent_1', + environment: 'env_1', + userMessage: 'hi', + ...params, + } as ManagedAgentRunSessionParams) + +beforeEach(() => { + vi.clearAllMocks() + runManagedAgentSession.mockResolvedValue({ + ok: true, + content: 'hello', + sessionId: 'sess_1', + inputTokens: 12, + outputTokens: 3, + }) +}) + +describe('managedAgentRunSessionTool.directExecution', () => { + it('errors when no credential key was injected', async () => { + const res = await run({ accessToken: undefined }) + expect(res.success).toBe(false) + expect(runManagedAgentSession).not.toHaveBeenCalled() + }) + + it('errors when agent or environment is blank', async () => { + const res = await run({ agent: ' ' }) + expect(res.success).toBe(false) + expect(runManagedAgentSession).not.toHaveBeenCalled() + }) + + it('fails closed when vaults are selected without acknowledgement', async () => { + const res = await run({ vaults: ['vlt_1'], vaultsAck: false }) + expect(res.success).toBe(false) + expect(res.error).toContain('Vault authorization') + expect(runManagedAgentSession).not.toHaveBeenCalled() + }) + + it('passes vaults through once acknowledged', async () => { + await run({ vaults: ['vlt_1'], vaultsAck: true }) + expect(runManagedAgentSession).toHaveBeenCalledWith( + expect.objectContaining({ apiKey: 'sk-ant-fake', vaultIds: ['vlt_1'] }) + ) + }) + + it('maps a successful run to content + sessionId + token usage', async () => { + const res = await run({}) + expect(res).toEqual({ + success: true, + output: { content: 'hello', sessionId: 'sess_1', inputTokens: 12, outputTokens: 3 }, + }) + }) + + it('surfaces a failed run as an error while preserving partial content', async () => { + runManagedAgentSession.mockResolvedValue({ ok: false, content: 'partial', error: 'boom' }) + const res = await run({}) + expect(res.success).toBe(false) + expect(res.error).toBe('boom') + expect(res.output.content).toBe('partial') + }) +}) diff --git a/apps/sim/tools/managed_agent/run_session.ts b/apps/sim/tools/managed_agent/run_session.ts new file mode 100644 index 00000000000..32e49336778 --- /dev/null +++ b/apps/sim/tools/managed_agent/run_session.ts @@ -0,0 +1,228 @@ +import { runManagedAgentSession } from '@/lib/managed-agents/run-session' +import { + isTruthyAck, + normalizeFiles, + normalizeMemoryAccess, + normalizeSessionParameters, + normalizeStringList, +} from '@/tools/managed_agent/normalizers' +import type { + ManagedAgentRunSessionParams, + ManagedAgentRunSessionResponse, +} from '@/tools/managed_agent/types' +import type { ToolConfig } from '@/tools/types' + +/** + * Opens a Claude Platform Managed Agent session and returns the assistant + * response as text. + * + * The block's `credential` picker supplies a Claude Platform service-account + * credential; the executor resolves it to the workspace API key and injects + * `accessToken` before `directExecution` runs. The session lifecycle + * (`runManagedAgentSession`) is pure `fetch` with no server-only deps, so the + * tool module stays safe to import from the client registry. + */ +export const managedAgentRunSessionTool: ToolConfig< + ManagedAgentRunSessionParams, + ManagedAgentRunSessionResponse +> = { + id: 'managed_agent_run_session', + name: 'Managed Agent Run Session', + description: + 'Open a Claude Platform Managed Agent session and return the assistant response as text.', + version: '1.0.0', + + params: { + credential: { + type: 'string', + required: true, + visibility: 'user-only', + description: + 'Claude Platform credential (Anthropic workspace API key) to run the agent with.', + }, + accessToken: { + type: 'string', + required: false, + visibility: 'hidden', + description: 'Workspace API key injected by the executor from the selected credential.', + }, + agent: { + type: 'string', + required: true, + visibility: 'user-only', + description: 'Managed-agent id inside the linked Claude workspace.', + }, + environment: { + type: 'string', + required: true, + visibility: 'user-only', + description: 'Environment id inside the linked Claude workspace.', + }, + environmentType: { + type: 'string', + required: false, + visibility: 'user-only', + description: + "Environment execution model hint ('cloud' | 'self_hosted'); the actual type is re-resolved server-side for routing.", + }, + userMessage: { + type: 'string', + required: true, + visibility: 'user-or-llm', + description: 'The user message to send to the Managed Agent.', + }, + vaults: { + type: 'array', + required: false, + visibility: 'user-only', + description: 'Zero or more vault ids for MCP tool auth.', + }, + vaultsAck: { + type: 'boolean', + required: false, + visibility: 'user-only', + description: 'Acknowledgement that the author may use the attached vaults.', + }, + memoryStoreId: { + type: 'string', + required: false, + visibility: 'user-only', + description: 'Optional Agent Memory Store id.', + }, + memoryAccess: { + type: 'string', + required: false, + visibility: 'user-only', + description: "Memory store access mode: 'read_write' (default) or 'read_only'.", + }, + memoryInstructions: { + type: 'string', + required: false, + visibility: 'user-only', + description: 'Per-attachment guidance for how the agent should use the memory store.', + }, + files: { + type: 'array', + required: false, + visibility: 'user-only', + description: 'File attachments (cloud envs only), as [{fileId, mountPath?}].', + }, + sessionParameters: { + type: 'object', + required: false, + visibility: 'user-only', + description: 'Key/value session metadata forwarded to the session.', + }, + }, + + // Unused: `directExecution` runs the session and short-circuits the HTTP + // path, but `ToolConfig` requires a `request` shape. + request: { + url: () => '', + method: 'POST', + headers: () => ({}), + }, + + directExecution: async (params, signal): Promise => { + const apiKey = params.accessToken + if (!apiKey) { + return { + success: false, + output: { content: '', sessionId: '' }, + error: 'No Claude Platform credential is selected, or it could not be resolved.', + } + } + + const agentId = params.agent?.trim() + const environmentId = params.environment?.trim() + if (!agentId || !environmentId) { + return { + success: false, + output: { content: '', sessionId: '' }, + error: 'An agent and an environment are required.', + } + } + + const vaultIds = normalizeStringList(params.vaults) + if (vaultIds.length > 0 && !isTruthyAck(params.vaultsAck)) { + return { + success: false, + output: { content: '', sessionId: '' }, + error: + 'Vault authorization is required — check the "I am authorized to use these vaults" acknowledgement on the block, or remove the selected vault(s).', + } + } + + const files = normalizeFiles(params.files) + const sessionParameters = normalizeSessionParameters(params.sessionParameters) + const memoryStoreId = params.memoryStoreId?.trim() || undefined + const memoryAccess = normalizeMemoryAccess(params.memoryAccess) + const memoryInstructions = params.memoryInstructions?.trim() || undefined + + // Title the Anthropic session so it is traceable to its Sim workflow from + // the Claude Platform console. Only the workflow id is available in the + // client-safe execution context (names would require a DB lookup). + const workflowId = params._context?.workflowId?.trim() + const title = workflowId ? `Sim workflow ${workflowId}` : undefined + + const environmentType = + params.environmentType === 'self_hosted' || params.environmentType === 'cloud' + ? params.environmentType + : undefined + + const result = await runManagedAgentSession({ + apiKey, + agentId, + environmentId, + userMessage: (params.userMessage ?? '').toString(), + ...(environmentType ? { environmentType } : {}), + ...(title ? { title } : {}), + ...(vaultIds.length > 0 ? { vaultIds } : {}), + ...(memoryStoreId ? { memoryStoreId } : {}), + ...(memoryStoreId && memoryAccess ? { memoryAccess } : {}), + ...(memoryStoreId && memoryInstructions ? { memoryInstructions } : {}), + ...(files.length > 0 ? { files } : {}), + ...(sessionParameters ? { sessionParameters } : {}), + ...(signal ? { signal } : {}), + }) + + if (!result.ok) { + return { + success: false, + output: { content: result.content, sessionId: result.sessionId ?? '' }, + error: result.error ?? 'Managed Agent session failed', + } + } + + return { + success: true, + output: { + content: result.content, + sessionId: result.sessionId ?? '', + ...(result.inputTokens !== undefined ? { inputTokens: result.inputTokens } : {}), + ...(result.outputTokens !== undefined ? { outputTokens: result.outputTokens } : {}), + }, + } + }, + + outputs: { + content: { + type: 'string', + description: 'Final assistant text from the Managed Agent session.', + }, + sessionId: { + type: 'string', + description: 'Anthropic session id (for logs / linking).', + }, + inputTokens: { + type: 'number', + description: 'Cumulative input tokens for the session.', + optional: true, + }, + outputTokens: { + type: 'number', + description: 'Cumulative output tokens for the session.', + optional: true, + }, + }, +} diff --git a/apps/sim/tools/managed_agent/types.ts b/apps/sim/tools/managed_agent/types.ts new file mode 100644 index 00000000000..39dd27be591 --- /dev/null +++ b/apps/sim/tools/managed_agent/types.ts @@ -0,0 +1,51 @@ +import type { ToolResponse } from '@/tools/types' + +/** + * Params accepted by the `managed_agent_run_session` tool. Values come from + * the Managed Agent block's subblocks in their raw runtime shapes; the tool's + * `directExecution` normalizes them before running the session. `accessToken` + * is injected by the executor from the selected `credential`. + */ +export interface ManagedAgentRunSessionParams { + /** Claude Platform service-account credential id (block picker value). */ + credential: string + /** Workspace API key injected by the executor from `credential` at run time. */ + accessToken?: string + /** Managed-agent id from the linked Claude workspace. */ + agent: string + /** Environment id from the linked Claude workspace. */ + environment: string + /** Env-type hint ('cloud' | 'self_hosted') from the block; re-resolved server-side. */ + environmentType?: string + /** The user's turn as plain text. Resolved by the executor. */ + userMessage: string + /** Zero or more vault ids for MCP auth (array, json string, or comma-list). */ + vaults?: unknown + /** Acknowledgement that the author may use the attached vaults. */ + vaultsAck?: boolean | string + /** Optional Agent Memory Store id. */ + memoryStoreId?: string + /** Memory store access mode — `read_write` (default) or `read_only`. */ + memoryAccess?: string + /** Per-attachment guidance for how the agent should use the memory store. */ + memoryInstructions?: string + /** Files-API files (cloud environments), as table rows, an array, or a comma list. */ + files?: unknown + /** Key/value session metadata, as table rows or a flat object. */ + sessionParameters?: unknown + /** Execution context injected by the executor (used to title the session for traceability). */ + _context?: { workflowId?: string } +} + +export interface ManagedAgentRunSessionResponse extends ToolResponse { + output: { + /** Final assistant text from the Managed Agent session. */ + content: string + /** Anthropic session id (for logs / linking). */ + sessionId: string + /** Cumulative input tokens for the session, when available. */ + inputTokens?: number + /** Cumulative output tokens for the session, when available. */ + outputTokens?: number + } +} diff --git a/apps/sim/tools/registry.ts b/apps/sim/tools/registry.ts index 61eb2f74dd1..72faeb561eb 100644 --- a/apps/sim/tools/registry.ts +++ b/apps/sim/tools/registry.ts @@ -2466,6 +2466,7 @@ import { mailgunListMessagesTool, mailgunSendMessageTool, } from '@/tools/mailgun' +import { managedAgentRunSessionTool } from '@/tools/managed_agent' import { mem0AddMemoriesTool, mem0GetMemoriesTool, mem0SearchMemoriesTool } from '@/tools/mem0' import { memoryAddTool, memoryDeleteTool, memoryGetAllTool, memoryGetTool } from '@/tools/memory' import { @@ -5486,6 +5487,7 @@ export const tools: Record = { mailgun_add_list_member: mailgunAddListMemberTool, mailgun_list_domains: mailgunListDomainsTool, mailgun_get_domain: mailgunGetDomainTool, + managed_agent_run_session: managedAgentRunSessionTool, sms_send: smsSendTool, jira_retrieve: jiraRetrieveTool, jira_update: jiraUpdateTool, diff --git a/apps/sim/tools/types.ts b/apps/sim/tools/types.ts index 261ac103a50..3d030303d91 100644 --- a/apps/sim/tools/types.ts +++ b/apps/sim/tools/types.ts @@ -183,8 +183,10 @@ export interface ToolConfig

{ /** * Direct execution function for tools that don't need HTTP requests. * If provided, this will be called instead of making an HTTP request. + * Receives the workflow execution's abort signal (when one is active) so + * long-running direct executions can propagate cancellation. */ - directExecution?: (params: P) => Promise + directExecution?: (params: P, signal?: AbortSignal) => Promise /** * Optional dynamic schema enrichment for specific params. diff --git a/scripts/check-api-validation-contracts.ts b/scripts/check-api-validation-contracts.ts index 6e44238b79f..dad531101fb 100644 --- a/scripts/check-api-validation-contracts.ts +++ b/scripts/check-api-validation-contracts.ts @@ -9,8 +9,8 @@ const QUERY_HOOKS_DIR = path.join(ROOT, 'apps/sim/hooks/queries') const SELECTOR_HOOKS_DIR = path.join(ROOT, 'apps/sim/hooks/selectors') const BASELINE = { - totalRoutes: 964, - zodRoutes: 964, + totalRoutes: 965, + zodRoutes: 965, nonZodRoutes: 0, } as const