Agent Manager

What is it?

A platform to create, run, and interact with agents.

You build in a drag-and-drop canvas, wiring agents and tools together into different orchestration patterns — workflow, hub-and-spoke, or swarm — using bidirectional connections.

Four layers of abstraction

What you build stacks up through four layers, each one a grouping of the layer below it:

Tools

Python functions. Either assigned to a utility agent, or dropped on the canvas on their own as a static workflow node.

Utility agent

A ReActReAct = Reason + Act. The agent reasons about its input, then decides whether to act (call a tool) before producing an output. agent with a system prompt. It takes an input, reasons about it, decides whether to use a tool (if it has one assigned), and returns an output.

Super agent

An abstraction — simply a grouping of utility agents and/or tools.

Orchestrator agent

An abstraction — a grouping of super agents, utility agents, tools, or other orchestrator agents. Context is shared across everything inside it.

Visually, the layers nest. An orchestrator agent groups super agents, utility agents, and tools (and even other orchestrator agents); expand any super agent and you'll find the same building blocks inside — utility agents that each may own one or more tools:

Technical FYI

Behind the scenes, the agentic orchestration is built on the Microsoft Agent FrameworkOne of the many pythonic, agentic frameworks on the market. Similar to LangGraph, CrewAI, Pydantic, etc., with the Agent Manager product team's own plethora of custom classes and integrations to make it work the way it does in the drag-and-drop canvas.

Conversational UI (CUI)

What is it?

A conversational chat interface — like ChatGPT or Claude.

Out of the box it's just chat; it isn't agenticAble to take actions to complete a task — reason and use tools — not just reply with text.. It becomes agentic by your design — you decide what it can do.

You do that by building a PersonaIn CUI, a Persona is an agent — a chat personality you configure with a system prompt and, optionally, tools.. A Persona is another word for agent.

Two things turn it from a plain chatbot into one that performs tasks and integrates with systems:

How it works

1. Create a Persona.
2. Give it a system prompt.
3. Write Python tools and assign them to the Persona.
4. Chat with it. The Persona carries out what you ask — following its system prompt and using the tools you gave it.

Integration

Which do I use?

Agent Manager and CUI are two separate applications. Each runs on its own — you might use one, the other, or both, depending on the use case.

Here's my rule of thumb for deciding. It's not prescriptive — just a general guideline from experience:

How does integration work?

There's no one-click button or setup page that wires them together. You connect them with API calls inside your Python tools. The pattern looks like this:

• A Persona in CUI has a tool that triggers a Super Agent or Orchestrator Agent in Agent Manager.
• That agent runs the tasks it's built to do. Along the way, an agent (or a standalone tool) calls CUI's API to push a message back into the same conversation thread you triggered it from.
• Agent Manager can pause its run and wait for a reply.
• A tool in CUI calls Agent Manager's API to send a response back into that run — its Huddle TraceAgent Manager's term for a single agent execution (run). — which then picks up where it left off.

So the communication is entirely API-based. Where and when you place these checkpoints is up to your design and flow.

Here's one example of that back-and-forth — a user asking for a security scan, then for fixes:

What's the implication of this?

There's no shared context between the LLMs in CUI and the LLMs in Agent Manager. Whatever the agents do in Agent Manager, the agents in CUI don't know about — and vice versa. They work in isolation. The two only ever meet at the communication step, where a small chunk of context is passed across in the API call.

Only the dashed chunks ever cross. Purple is the context handed over when CUI triggers Agent Manager and when it returns; orange is messages Agent Manager emits back to CUI while it's still running. Everything else in each window stays private.

Utility Agents

A utility agent is a ReActReAct = Reason + Act. The agent reasons about its input, then decides whether to act (call a tool) before producing an output. agent — an LLM node that takes an input, reasons about it, and decides whether to use a tool before answering. You configure it with a system prompt (how it should behave) and, optionally, one or more tools it can call.

Example: a User Story Creator

Let's build one. Create a utility agent and fill in the fields below — use the copy button on each to paste into the interface in Agent Manager. Leave Tags and Tools alone for now.

Save the utility agent, then move on to the next section to create a super agent that contains this User Story Creator — so you can see it in action.

Super Agents

To test a utility agent, you put it inside a super agent and run that.

Create a new super agent, search for your User Story Creator in the Utility Agents panel, then drag and drop it onto the Flow canvas. Connect the Start and End nodes to it.

Save the super agent, then go back to the main Super Agents page. Search for the super agent you created and click the play button.

Enter your input — for example, "Create me a user story about a personal banking application" — then click Run.

After clicking Run, you're taken to the Huddle TraceAgent Manager's term for a single agent execution (run). page for the execution, where you can monitor the activity of the agent(s). Move on to the next section to learn more about Huddle Traces.

Huddle Traces

This is where you monitor the granular activity of a super or orchestrator agent — the inputs, tool uses, and outputs of every agent and standalone tool that runs during an execution.

Once the agent starts running, click the individual nodes in the swimlane view to see what's happening.

Tool executions show the final return of the Python function as Result, plus Logs — anything you emit from the function with print statements.

For example, clicking the fetch_org_docs node from the trace above reveals its details:

Tools

Tools are Python functions an agent (or a standalone canvas node) can call to actually do something — fetch data, hit an API, run logic. You give the tool a name, describe its parameters, and write the function; the agent decides when to run it.

Here's the example fetch_org_docs tool from the trace in the previous section, which is a dummy function that returns a static string:

import core

def fetch_org_docs(limit: int) -> str:
    """Fetch up to `limit` org docs from the knowledge base."""

    print(f"Fetching document from Confluence with limit={limit}…")

    doc = (
        "ACME Inc. uses the agile methodology for capturing product requirements from"
        "the business — breaking them into epics and user stories, managed in Jira "
        "and refined each sprint. All User Stories have `Made by ACME Inc` at the end."
    )

    return doc

core.output(fetch_org_docs(**core.args()))

Two lines are required in every tool:

1. import core at the top of the tool.
2. core.output(<FUNCTION_NAME>(**core.args())) on the last line of the tool — that's what feeds the agent's arguments into your function and returns the result.

Next, save the tool, then go back to the User Story Creator utility agent and add the fetch_org_docs tool to it.

Replace the Utility Agent's Detail (system prompt) with this:

You can now re-run the super agent and monitor the Huddle Trace to see the utility agent use the tool!

Orchestrator Agents

An orchestrator agentAn abstraction — a grouping of super agents, utility agents, tools, or other orchestrator agents. Context is shared across everything inside it. is just one level of abstraction above a super agent. It lets you group everything you've learned so far — toolsPython functions. Either assigned to a utility agent, or dropped on the canvas on their own as a static workflow node., utility agentsA ReAct agent with a system prompt. It takes an input, reasons about it, decides whether to use a tool (if it has one assigned), and returns an output., super agentsAn abstraction — a grouping of utility agents and/or tools., and even other orchestrator agentsAn abstraction — a grouping of super agents, utility agents, tools, or other orchestrator agents. Context is shared across everything inside it. — all within one agent.

So far, we've made a User Story super agent with a sample tool.

Now, your first exercise is to create a super agent that creates test plans.

Before we create an orchestrator agent, we need to create one more utility agent — the one that will act as our orchestrator.

Now we can assemble everything we've made into one orchestrator agent. Navigate to the Orchestrator page and assemble it like this:

Notice the bidirectional connections between the central utility agent and the adjacent super agents.

When done, save the agent and run it with this prompt:

Following the Huddle Trace, you'll see the trace-wise flow and activity of the agents — especially the orchestrator. You can also click the blue super agent nodes in the swimlane diagram to view their individual nested traces.

CUI Integration

What do we have so far?

So far we've learned how to trigger Super and Orchestrator Agents directly in Agent Manager. We also heard about CUI being the user-facing interface for chat-based use cases. So how do we integrate it with agents? Recall the Integration page, where we connect the two with API calls inside Python functions.

Trigger Agent from CUI

We need a PersonaIn CUI, a Persona is an agent — a chat personality you configure with a system prompt and, optionally, tools. plus a tool in CUI that triggers our Business Orchestrator Agent with a text input.

Create a tool in CUI as follows — use the copy button on each field to paste into the interface:

import asyncio
import os
from uuid import UUID
import httpx

# ------------ CUI CREDENTIALS ------------
USERNAME = config_service.get("tool-cui-user","")
PASSWORD = config_service.get("tool-cui-password","")
BASE_URL = config_service.get("tool-cui-base-url","")

# ----------- CUI RELATED DATA ------------
ATTACHMENTS = user_query_context.attachments
CONVERSATION_ID = user_query_context.conversation_id  # ID of the conversation
JWT_TOKEN = user_query_context.user_specific_jwt_token # JWT token for authentication

# ------------ AM CREDENTIALS ------------
API_URL = config_service.get("agentmanager-api-url","")
API_KEY = config_service.get("agentmanager-api-key","")

# ------------ AM AGENT ID AND TYPE ------------
RUN_AGENT_TEAM_ID = "c95edc15-ad49-4695-a006-a5113377ac90"
AGENT_TYPE = "ORCHESTRATOR" # Super Agent = "TEAM", Orchestrator Agent = "ORCHESTRATOR"
# EDIT THESE ^

# ------------ INPUT SENT TO AGENT MANAGER ------------
TASK = {
    "USER_INPUT": USER_INPUT,
    "cui_conversation_id": CONVERSATION_ID,
    "user_id": user_query_context.username,
    "recipient_id": user_query_context.user_id,
    "network_id": user_query_context.external_network_run_id
}

def normalize_filename(filename):
    name, ext = os.path.splitext(filename)
    inner_name, inner_ext = os.path.splitext(name)
    if inner_ext:
        name = f"{inner_name}_{inner_ext.lstrip('.')}"
    name = name.replace(".", "_").replace(" ", "_")
    suffix = os.urandom(2).hex()
    return f"{name}_{suffix}{ext}"

async def get_attachment_content(attachment_id):
	from app.dependencies import get_attachments_service
	from app.utils.dependencies import inject
	from uuid import UUID

	print(f"Start getting content of Attachment id: {attachment_id}")
	attachments_service = await inject(get_attachments_service)
	(attachment_info, data) = await attachments_service.get_attachment(
		file_id=UUID(attachment_id), user_id=user_query_context.username
	)
	print(f"#############Attachment info: {attachment_info}")
	return data

async def get_presigned_url(filename):
    async with httpx.AsyncClient(timeout=30) as client:
        payload = {
            "query": """mutation SetUploadedArtifact($uploadedArtifact: UploadedArtifactInput!, $setUploadedArtifactId: String) {
                        setUploadedArtifact(
                            uploadedArtifact: $uploadedArtifact
                            id: $setUploadedArtifactId
                        ) {
                            uploadedArtifact {
                                id
                                name
                                description
                                createdBy
                                createdOn
                                updatedBy
                                updatedOn
                                s3Key
                                tags
                                __typename
                            }
                            message
                            presignedUrl
                            __typename
                        }
                    }""",
            "variables": {
                "uploadedArtifact": {
                    "name": filename,
                    "description": "User Input File"
                }
            },
        }
        headers = {"Content-Type": "application/json", "x-api-key": API_KEY}

        try:
            response = await client.post(API_URL, json=payload, headers=headers)
            print(
                f"Presigned URL response status code: {response.status_code}")
            print(f"Presigned URL response content: {response.text}")

            response.raise_for_status()
            data = response.json()

            # Enhanced error handling and debugging
            if "errors" in data:
                print(f"GraphQL errors: {data['errors']}")
                raise ValueError(f"GraphQL errors: {data['errors']}")

            # Check if the response has the expected structure
            if "data" not in data:
                print(f"Missing 'data' field in response: {data}")
                raise ValueError(f"Unexpected response structure: {data}")

            if data["data"] is None:
                print(f"GraphQL returned null data: {data}")
                raise ValueError(f"GraphQL returned null data: {data}")

            if "setUploadedArtifact" not in data["data"]:
                print(f"Missing setUploadedArtifact in response: {data}")
                raise ValueError(
                    f"Missing setUploadedArtifact in response: {data}")

            set_uploaded_artifact = data["data"]["setUploadedArtifact"]
            if set_uploaded_artifact is None:
                print(f"setUploadedArtifact is null: {data}")
                raise ValueError(f"setUploadedArtifact is null")

            if "presignedUrl" not in set_uploaded_artifact:
                print(
                    f"Missing presignedUrl in setUploadedArtifact: {set_uploaded_artifact}")
                raise ValueError(f"Missing presignedUrl in response")

            presigned_url = set_uploaded_artifact["presignedUrl"]
            if presigned_url is None:
                print(f"presignedUrl is null: {set_uploaded_artifact}")
                raise ValueError(f"presignedUrl is null")

            return presigned_url

        except httpx.HTTPStatusError as e:
            print(f"HTTP error getting presigned URL: {e}")
            print(f"Response content: {e.response.text}")
            raise
        except Exception as e:
            print(f"Error getting presigned URL: {e}")
            raise

async def upload_file(presigned_url, file_content, content_type):
    if not presigned_url:
        raise ValueError("presigned_url is None or empty")

    async with httpx.AsyncClient(timeout=60) as client:
        headers = {"Content-Type": content_type}
        try:
            print(f"Uploading file to: {presigned_url}")
            print(f"File size: {len(file_content)} bytes")
            print(f"Content type: {content_type}")

            response = await client.put(
                presigned_url, headers=headers, content=file_content
            )
            print(f"Upload response status code: {response.status_code}")
            print(f"Upload response text: {response.text}")

            response.raise_for_status()
            return response.status_code

        except httpx.HTTPStatusError as e:
            print(f"HTTP error uploading file: {e}")
            print(f"Response content: {e.response.text}")
            raise
        except Exception as e:
            print(f"Error uploading file: {e}")
            raise

async def trigger_execution():
    async with httpx.AsyncClient(timeout=30) as client:
        payload = {
            "query": """mutation RunAgent($runAgentId: String!, $type: RunAgentType!, $task: String!) {
                runAgent(id: $runAgentId, type: $type, task: $task) {
                    message
                    traceId
                    __typename
                }
            }""",
            "variables": {"runAgentId": RUN_AGENT_TEAM_ID, "task": str(TASK), "type": AGENT_TYPE},
        }
        print(f"Payload for execution: {payload}")
        headers = {"Content-Type": "application/json", "x-api-key": API_KEY}
        try:
            response = await client.post(API_URL, json=payload, headers=headers)
            print(f"Execution trigger response status: {response.status_code}")
            print(f"Execution trigger response: {response.text}")
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"Error triggering execution: {e}")
            return {"message": "Failed to trigger execution", "error": str(e)}


async def main():
    presigned_url = None
    
    headers = {
        "accept": "application/json",
        "apiToken": JWT_TOKEN,
        "Content-Type": "application/json",
    }

    csv_counter = 1
    docx_counter = 1
    pdf_counter = 1
    zip_counter = 1
    pptx_counter = 1
    md_counter = 1
    json_counter = 1
    txt_counter = 1
    
    for attachment in ATTACHMENTS:
        if attachment.filename.endswith('.zip'):
            print(f"Processing code repository: {attachment.filename}")
            zipkey = f"zip{zip_counter}"
            TASK[zipkey] = attachment.filename
            print(f"Uploading file {zipkey} (zip)...")
            
            try:
                text_content = await get_attachment_content(attachment.id)
				# text_content = await get_attachment_content(attachment.id, headers=headers)
                
                print(f"Start retrieving PRESIGNED_URL using {attachment.filename}")
                presigned_url = await get_presigned_url(normalize_filename(attachment.filename))
                print(f"PRESIGNED_URL: {presigned_url}")
                
                upload_status = await upload_file(
                    presigned_url,
                    text_content,
                    "application/zip",
                )
                print(f"File upload status: {upload_status}")
                zip_counter += 1  # Increment counter after successful processing
            except Exception as e:
                print(f"Error processing zip file: {e}")
                raise

        elif attachment.filename.endswith('.csv'):
            print(f"Processing CSV file: {attachment.filename}")
            csv_key = f"csv{csv_counter}"
            TASK[csv_key] = attachment.filename
            print(f"Uploading {csv_key}...")
            try:
                text_content = await get_attachment_content(attachment.id)
				# text_content = await get_attachment_content(attachment.id, headers=headers)
                print(f"Start retrieving PRESIGNED_URL using {attachment.filename}")
                presigned_url = await get_presigned_url(normalize_filename(attachment.filename))
                print(f"PRESIGNED_URL: {presigned_url}")
                upload_status = await upload_file(
                    presigned_url,
                    text_content,
                    "text/csv",
                )
                print(f"File upload status: {upload_status}")
                csv_counter += 1  # Increment counter after successful processing
            except Exception as e:
                print(f"Error processing CSV file: {e}")
                raise

        elif attachment.filename.endswith('.docx'):
            print(f"Processing DOCX file: {attachment.filename}")
            docx_key = f"docx{docx_counter}"
            TASK[docx_key] = attachment.filename
            print(f"Uploading {docx_key}...")
            try:
                text_content = await get_attachment_content(attachment.id)
				# text_content = await get_attachment_content(attachment.id, headers=headers)
                print(f"Start retrieving PRESIGNED_URL using {attachment.filename}")
                presigned_url = await get_presigned_url(normalize_filename(attachment.filename))
                print(f"PRESIGNED_URL: {presigned_url}")
                upload_status = await upload_file(
                    presigned_url,
                    text_content,
                    "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
                )
                print(f"File upload status: {upload_status}")
                docx_counter += 1
            except Exception as e:
                print(f"Error processing DOCX file: {e}")
                raise

        elif attachment.filename.endswith('.pdf'):
            print(f"Processing PDF file: {attachment.filename}")
            pdf_key = f"pdf{pdf_counter}"
            TASK[pdf_key] = attachment.filename
            print(f"Uploading {pdf_key}...")
            try:
                text_content = await get_attachment_content(attachment.id)
				# text_content = await get_attachment_content(attachment.id, headers=headers)
                print(f"Start retrieving PRESIGNED_URL using {attachment.filename}")
                presigned_url = await get_presigned_url(normalize_filename(attachment.filename))
                print(f"PRESIGNED_URL: {presigned_url}")
                upload_status = await upload_file(
                    presigned_url,
                    text_content,
                    "application/pdf",
                )
                print(f"File upload status: {upload_status}")
                pdf_counter += 1
            except Exception as e:
                print(f"Error processing PDF file: {e}")
                raise

        elif attachment.filename.endswith('.pptx'):
            print(f"Processing PPTX file: {attachment.filename}")
            pptx_key = f"pptx{pptx_counter}"
            TASK[pptx_key] = attachment.filename
            print(f"Uploading {pptx_key}...")
            try:
                text_content = await get_attachment_content(attachment.id)
                print(f"Start retrieving PRESIGNED_URL using {attachment.filename}")
                presigned_url = await get_presigned_url(normalize_filename(attachment.filename))
                print(f"PRESIGNED_URL: {presigned_url}")
                upload_status = await upload_file(
                    presigned_url,
                    text_content,
                    "application/vnd.openxmlformats-officedocument.presentationml.presentation",
                )
                print(f"File upload status: {upload_status}")
                pptx_counter += 1
            except Exception as e:
                print(f"Error processing PPTX file: {e}")
                raise

        elif attachment.filename.endswith('.md'):
            print(f"Processing MD file: {attachment.filename}")
            md_key = f"md{md_counter}"
            TASK[md_key] = attachment.filename
            print(f"Uploading {md_key}...")
            try:
                text_content = await get_attachment_content(attachment.id)
                print(f"Start retrieving PRESIGNED_URL using {attachment.filename}")
                presigned_url = await get_presigned_url(normalize_filename(attachment.filename))
                print(f"PRESIGNED_URL: {presigned_url}")
                upload_status = await upload_file(
                    presigned_url,
                    text_content,
                    "text/markdown",
                )
                print(f"File upload status: {upload_status}")
                md_counter += 1
            except Exception as e:
                print(f"Error processing MD file: {e}")
                raise

        elif attachment.filename.endswith('.json'):
            print(f"Processing JSON file: {attachment.filename}")
            json_key = f"json{json_counter}"
            TASK[json_key] = attachment.filename
            print(f"Uploading {json_key}...")
            try:
                text_content = await get_attachment_content(attachment.id)
                print(f"Start retrieving PRESIGNED_URL using {attachment.filename}")
                presigned_url = await get_presigned_url(normalize_filename(attachment.filename))
                print(f"PRESIGNED_URL: {presigned_url}")
                upload_status = await upload_file(
                    presigned_url,
                    text_content,
                    "application/json",
                )
                print(f"File upload status: {upload_status}")
                json_counter += 1
            except Exception as e:
                print(f"Error processing JSON file: {e}")
                raise

        elif attachment.filename.endswith('.txt'):
            print(f"Processing TXT file: {attachment.filename}")
            txt_key = f"txt{txt_counter}"
            TASK[txt_key] = attachment.filename
            print(f"Uploading {txt_key}...")
            try:
                text_content = await get_attachment_content(attachment.id)
                print(f"Start retrieving PRESIGNED_URL using {attachment.filename}")
                presigned_url = await get_presigned_url(normalize_filename(attachment.filename))
                print(f"PRESIGNED_URL: {presigned_url}")
                upload_status = await upload_file(
                    presigned_url,
                    text_content,
                    "text/plain",
                )
                print(f"File upload status: {upload_status}")
                txt_counter += 1
            except Exception as e:
                print(f"Error processing TXT file: {e}")
                raise

    print("Triggering execution...")
    execution_response = await trigger_execution()
    return f"Execution response: {execution_response}"

try:
    result = await main()
    return str(result)
except Exception as e:
    return f"Exception occurred: {e}"

Before saving, modify line 21 — replace RUN_AGENT_TEAM_ID with the ID of the Orchestrator Agent you created earlier.

The script makes a single API call to Agent Manager's run endpoint, passing the user's input as the agent's input.

Save the tool. Then create a Persona, give it a system prompt that tells it to call business_orch_agent_tool with the user's request, and add this tool to it. Now when you chat with the Persona, it triggers the Business Orchestrator Agent in Agent Manager.

Send Message from AM to CUI

Store CUI Variables

To send messages from AM back to CUI, we first need to store, on the AM side, some of the variables sent from CUI to AM (cui_conversation_id, user_id, etc.).

Do this by creating a new tool and Utility Agent that store these variables at runtime, before anything else proceeds.

import typing
import core

def save_local_variable(key: typing.Annotated[str, "The variable name/key to save"] = None,
    value: typing.Annotated[str, "The value to save"] = None):
    """
    Saves a local variable by key.
    """
    if value is None:
        return "Error: 'value' must be provided."
    try:
        existing = core.get_local_value(key)
        if existing is not None:
            return f"Variable '{key}' already exists with value '{existing}'. Not overwritten."
    except Exception:
        pass
    core.set_local_value(key, value)
    return f"Saved: {key} = {value}"

core.output(save_local_variable(**core.args()))

Then create a Utility Agent that uses this tool, with the system prompt below:

Then, connect this agent between the START node and the Business Orchestrator Agent.

Send Message to CUI

Next, we need a tool which will allow Utility Agents to send a message to CUI:

import requests
import core
import typing

def send_msg_to_cui(
    message: typing.Annotated[str, "The message content to send to CUI."] = None
):
    """Send a message to CUI on behalf of the agent.

    Args:
        message: The message content to send to CUI.

    Returns:
        The original message string in all cases.
    """

    try:
        # GET ATR CREDENTIALS
        credentials = core.get_secret(name="atr")
        url_base = credentials.get("url")
        # RETRIEVE STORED CUI IDS
        recipient_id = core.get_local_value("recipient_id")
        conversation_id = core.get_local_value("cui_conversation_id")

        url = f"{url_base}/atr-gateway/identity-management/api/v1/auth/short-token"
        headers = {"Content-Type": "application/json"}
        body = {"username": credentials.get("username"), "password": credentials.get("password")}
        response = requests.post(url, json=body, headers=headers)
        token = response.json()["token"]

        sys_response_id = ""
        headers = {"apiToken": token}
        url = f"{url_base}/atr-gateway/genai/messages"
        body = {
            "conversation_id": conversation_id,
            "content": message,
            "recipient_id": recipient_id,
            "push_message_metadata": {
                "system_name": "agent_manager",
                "system_awaiting_reply": False,
                "system_response_id": sys_response_id,
            },
        }

        response = requests.post(url, json=body, headers=headers)
        print(f"Sent message to CUI successfully. Response: {response.content}")
        return message

    except Exception as e:
        print(f"Error sending message to CUI: {e}")
        return message

core.output(send_msg_to_cui(**core.args()))

Now create a Utility Agent which will utilise this tool.

Now add and connect this agent to the Utility Agent inside BOTH the User Story Super Agent and Test Plan Super Agent. For example, the User Story Super Agent's flow becomes:

Jira MCP

In this lab, we'll learn how to use the Jira MCPHosted collection of reusable Python tools for agents. in Agent Manager to push data to a Jira instance.

First we'll add Jira-related credentials into Agent Manager, then create a tool to interface with the MCP, create an agent which utilises it, then incorporate it into our existing agent from the previous lab.

Create Secret

First, let's create the Jira secret.

Do this by navigating to the Secrets ManagerLocated in the top right of Agent Manager, under the settings button dropdown. page and clicking + Add New Secret.

Then, fill in the fields below using the credentials from the provided Excel:

Create Jira MCP Tool

Now we need a tool to talk to the hosted Jira MCP client. The client has its own LLM ReActReAct = Reason + Act. The agent reasons about its input, then decides whether to act (call a tool) before producing an output. agent behind the scenes, so the tool just hands it a prompt describing what to do in Jira, and the MCP client (LLM) decides which MCP tools to call to accomplish the task.

For example, the Utility Agent in Agent Manager is instructed to push User Stories to Jira. It calls the Jira MCP Tool with a prompt for the MCP Client; the MCP Client (an LLM ReAct agent underneath) interprets the prompt, then decides which Jira tool(s) in the MCP Server to use.

The result then travels back up the stack: the MCP Tool succeeds, the MCP Client agent acknowledges it and returns — which becomes the Agent Manager tool's output. The Utility Agent receives that and generates a response (e.g. "User stories successfully pushed").

import core

def jira_mcp_client_tool(prompt):
    prompt = [prompt]
    print(f"Calling Jira MCP Client with prompt: {prompt}")
    configuration = "OpenAI GPT-4.1"
    mcp_config = "group_1_jira_secret"

    result = core.mcp_jira_init(prompt, configuration, mcp_config)
    return result

core.output(jira_mcp_client_tool(**core.args()))

Create Jira MCP Agent

Now let's create the Utility Agent that uses this tool — it interprets the incoming request and calls jira_mcp_client_tool with a well-formed prompt.

Integrate with Agent

Now we'll integrate the Jira MCP Utility Agent into our Super Agent from the previous labs.

Connect to User Story Agent

Open the User Story Super Agent and drop the Jira Agent onto the Flow canvas, connecting it between the User Story Creator and the Send to CUI utility agents.

Now test the same agent from the Persona we made in CUI, with the following prompt.

Now you can monitor your Jira board and see the generated stories populate when the agent completes that step!

Human in the Loop

Sometimes we may want an Agent to pause and wait for user input partway through its run.

For example, once the user stories are generated, we might want the user to weigh in — are they good? Should they be regenerated? And if they're happy with them, is it good to push them to Jira?

We handle this with a human-in-the-loop step that pauses the agent's execution in Agent Manager and waits for input from the user in CUI before carrying on with the rest of the flow.

To do this, we need two pieces:

1. A wait step in Agent Manager (a utility agent and a tool) that pauses and waits for a reply.
2. A reply tool for the Persona in CUI that sends the user's answer back.

AM Component

If you recall from Integration, communication between Agent Manager and CUI is through API calls. Meaning, we'll need to:

1. Create a tool which will:
   1. Send a message to CUI,
   2. Pause the execution, and
   3. Wait for a reply from CUI.
2. Create a utility agent which will leverage the tool.

Start by creating this new tool:

import requests
import core
import typing

def ask_for_user_input_cui(
    message: typing.Annotated[str, "The message content to send to user."] = None
):
    """Send a message to CUI on behalf of the agent, then wait for the user's reply."""

    # Get CUI details from AM Secrets Manager
    credentials = core.get_secret(name="atr")
    url_base = credentials.get("url")
    recipient_id = core.get_local_value("recipient_id")
    conversation_id = core.get_local_value("cui_conversation_id")

    # CUI Auth
    auth_resp = requests.post(
        f"{url_base}/atr-gateway/identity-management/api/v1/auth/short-token",
        json={"username": credentials.get("username"), "password": credentials.get("password")},
        headers={"Content-Type": "application/json"},
        timeout=30,
    )
    auth_resp.raise_for_status()
    token = auth_resp.json()["token"]

    # Send message with reply enabled
    send_resp = requests.post(
        f"{url_base}/atr-gateway/genai/messages",
        json={
            "conversation_id": conversation_id,
            "content": message,
            "recipient_id": recipient_id,
            "push_message_metadata": {
                "system_name": "agent_manager",
                "system_awaiting_reply": True,
                "system_response_id": core.trace_id(),
            },
        },
        headers={"apiToken": token},
        timeout=30,
    )
    send_resp.raise_for_status()
    print(f"Sent message to CUI. Status: {send_resp.status_code}")

    # Wait for reply
    try:
        user_input = core.user_input(timeout=21600)  # Wait for user input for up to 6 hours (21600 seconds)
    except Exception as e:
        print(f"User input timeout: {e}")
        return "User Input Timeout - Handoff to END"

    return f"Received user input: {user_input}"

core.output(ask_for_user_input_cui(**core.args()))

Then create the utility agent:

When done, you can add this to the User Story Super Agent.

Now trigger the agent from CUI or Agent Manager itself to quickly test the User Input feature. Monitor the Huddle Trace and watch closely for the flow to reach the Approval Agent.

We can test the user input feature by replying directly in the Huddle Trace, before we integrate on the CUI side.

Once the Huddle Trace reaches the above state, click the most recent trace for the ask_for_user_input_cui tool. This shows a Give Input button you can click to enter a message to send to the agent. You can also click the Utility Agent trace that called this tool to view the message that would be sent to the user.

CUI Component

We now need to implement the user input functionality on the CUI side to integrate with Agent Manager. Recall how the two communicate via API — this means we need to:

1. Create a new tool in CUI that invokes a different Agent Manager endpoint, which sends the reply message back — just like we did manually in the Huddle Trace UI.
2. Add this tool to the existing Persona and adjust its system prompt on how to use the tool.

Start by creating the tool in CUI:

import httpx

from app.domain.tools.utils.context import context_util_helper
from app.models.canvas_document import CanvasDocument

API_URL = config_service.get("agentmanager-api-url","")
API_KEY = config_service.get("agentmanager-api-key","")

constructed_message = "User Reply: " + MESSAGE

async def reply_to_agent_manager(MESSAGE, TRACE_ID):
    headers = {
        "Content-Type": "application/json",
        "x-api-key": API_KEY,
    }
    data = {
        "query": """mutation SendUserResponse($traceId: String!, $message: String!) {
  sendUserResponse(traceId: $traceId, message: $message) {
    message
    __typename
  }
}""",
        "variables": {
            "traceId": TRACE_ID,
            "message": constructed_message,
        },
    }

    # Debug logging
    print(f"DEBUG - trace_id: {TRACE_ID}")
    print(f"DEBUG - message: {MESSAGE}")
    print(f"DEBUG - headers: {headers}")
    print(f"DEBUG - data: {data}")

    async with httpx.AsyncClient(timeout=20) as client:
        response = await client.post(API_URL, headers=headers, json=data)
        print(f"DEBUG - status_code: {response.status_code}")
        print(f"DEBUG - response headers: {dict(response.headers)}")
        print(f"DEBUG - response text: {response.text}")

        response.raise_for_status()

        response_json = response.json()
        print(f"DEBUG - response JSON: {response_json}")

        if "errors" in response_json:
            raise ValueError(f"GraphQL errors: {response_json['errors']}")

        return "Your message has been sent successfully to Agent Manager. Please wait while the Agent continues its tasks."

# Run the async function
if TRACE_ID is None:
    return "Trace ID is required to send a message to Agent Manager."
if not MESSAGE:
    return "Message is required to send a message to Agent Manager."

try:
    result = await reply_to_agent_manager(MESSAGE, TRACE_ID)
    return str(result)
except Exception as e:
    return f"Exception occurred: {e}"

Now navigate to your existing Persona, give it the new tool you made, and add this to the existing prompt:

And that's it!

Trigger the agent once again and monitor CUI for the message that gets sent from Agent Manager. You'll see it pop up looking like this:

Back in 2020/2021, we saw LLM chatbots like ChatGPT and Claude come to life — just chat interfaces for interacting with the underlying LLM.

In 2023, we got the ability to give these LLMs tools, letting them interact with external systems through APIs — this gave rise to the concept of an "agent".

In 2025, Claude Code took the world by storm and coined the term "agentic harness" — in a nutshell, it's what lets Claude Code browse files, run commands, and interact with code directly and autonomously. Codex, Antigravity CLI, and OpenCode are all similar agentic harnesses out there in the market.

What does this mean?

The possibilities are endless — Claude Code can read and write files, run code and scripts, and more. This opens the door to use cases such as:

• Forward engineering / code generation / upgrades
• Reverse engineering / documentation generation
• Change impact analysis
• And more…

Why is this relevant?

In Agent Manager, we have the ability to leverage Claude Code as a tool — meaning anything you can do locally with Claude Code, we can implement in Agent Manager as an automation.

Mesh Code

In Agent Manager, there is a pre-made function/tool called Mesh Code, which runs Claude Code underneath.

Start by creating the following tool:

import core
import os

def get_repository_dir() -> str:
    repository_dir = core.get_local_value("repository_dir")

    if not repository_dir:
        project_dir = "./project"
        core.set_local_value("project_dir", project_dir)

        repository_dir = "./project/repository"
        core.set_local_value("repository_dir", repository_dir)
        os.makedirs(repository_dir, exist_ok=True)

        # Create a .gitkeep file in the repository directory to ensure it's tracked
        gitkeep_path = os.path.join(repository_dir, ".gitkeep")
        with open(gitkeep_path, "w") as f:
            f.write("")  # Empty file

    repository_dir = os.path.expanduser(repository_dir)

    return repository_dir


result = core.run_mesh_code(
    repository_dir=get_repository_dir(),
    prompt=core.args().get("prompt"),
    secret_name="Bedrock Sonnet 4.5",
    new_conversation=False
)

core.output(result)

Behind the scenes, this is how it works in Agent Manager:

* Utility Agent is optional; you can set a static prompt for Mesh Code as a standalone tool in the Flow of a Super Agent.

The Mesh Code tool spawns an instance of Claude Code. Whatever you pass to the tool's prompt parameter goes straight to that instance, which then behaves just like local Claude Code. It runs on an ephemeral Linux filesystem that spins up alongside the agent execution.

We do this by having a Utility agent run a tool before Mesh Code, which will download and load a codebase or any files provided by the user into the Linux Filesystem, where Claude Code will be spawned.

This is what we'll cover in the next section before testing Mesh Code.

Reverse Engineering Agent

In this lab, we'll create a Super Agent that generates documentation for a provided codebase.

The end outcome we're aiming for is something like this:

* A static prompt is set in the run_mesh_code tool, so Claude always runs the same prompt for the agent, regardless of the codebase.

The Save Variables and Send to CUI Utility Agents we've already made in the previous labs, so we'll be reusing those.

We'll build this up over the next few sections — loading a codebase into the filesystem, running Mesh Code against it, then generating a downloadable URL for the result.

Load Codebase from Zip

In this section, we'll focus on loading a zipped codebase specifically for Mesh Code.

The CUI tool we implemented already supports uploading files to the Uploaded Artifacts tab in Agent Manager — we just need to load it into the filesystem, as mentioned in the previous section.

Start by creating the following tool, which loads an uploaded artifact — provided the file exists and the given filename matches:

import typing
import requests
import core
import io
import os
import zipfile

def dl_uploaded_artifact_for_mesh(
    filename: typing.Annotated[str, "The filename of the artifact to download. Zip files are extracted; all other file types are saved directly."] = None):
    """
    Download a uploaded artifact. Zip files are extracted into the repository directory;
    all other file types (CSV, PDF, DOCX, etc.) are saved directly to the repository directory.

    Args:
        filename: The filename of the uploaded artifact to download

    Returns:
        str: Success message or error description
    """
    import shutil

    if not filename:
        raise ValueError("Filename parameter is required and cannot be None or empty.")

    project_dir = "./project"
    core.set_local_value("project_dir", project_dir)

    repository_dir = "./project/repository"
    core.set_local_value("repository_dir", repository_dir)

    try:
        os.makedirs(repository_dir, exist_ok=True)
    except Exception as e:
        raise Exception(f"Failed to create directory {repository_dir}: {e}")

    try:
        file = core.get_artifact(name=filename, type='uploaded', text=False)
    except Exception as e:
        raise Exception(f"Failed to download artifact '{filename}': {e}")

    # If the file is a zip file, extract it; otherwise, save it directly
    if filename.lower().endswith('.zip'):
        try:
            with zipfile.ZipFile(io.BytesIO(file)) as zip_ref:
                zip_ref.extractall(repository_dir)

            macosx_path = os.path.join(repository_dir, "__MACOSX") # Remove remnant __MACOSX folder
            if os.path.exists(macosx_path):
                shutil.rmtree(macosx_path)
        except Exception as e:
            raise Exception(f"Failed to extract '{filename}' to {repository_dir}: {e}")

        return f"Successfully downloaded and extracted '{filename}' to {repository_dir}."

    # Not Zip file, save directly
    else:
        try:
            file_path = os.path.join(repository_dir, filename)
            with open(file_path, 'wb') as f:
                f.write(file)
        except Exception as e:
            raise Exception(f"Failed to save '{filename}' to {repository_dir}: {e}")

        return f"Successfully downloaded '{filename}' to {file_path}."

core.output(dl_uploaded_artifact_for_mesh(**core.args()))

Then create the Utility Agent:

Now move on to the next section.

Generate Artifact URL

Once Mesh (Claude) Code has generated code or documents, the Agent will need a way to send this to the end user. Large/not-purely-text artifacts like code need a downloadable URL to send to the user.

You can do this with the following tool, which will upload any generated artifacts in the filesystem to S3 and return a presigned URL — add it to Agent Manager:

import os
import uuid
import zipfile
import datetime
import core

def generate_artifacts_url():
    """Zip repository_dir, upload to S3, and return a presigned URL.

    Returns:
        A presigned S3 URL string on success, or an ERROR string on failure.
    """
    repository_dir = core.get_local_value("repository_dir")
    artifact_path = "full_artifact.zip"
    artifact_prefix = "artifacts/{artifact_id}"

    exclude_dirs = {"node_modules"}

    def zip_folder(folder_path, zip_file_name):
        if not os.path.isdir(folder_path):
            raise FileNotFoundError(f"The folder {folder_path} does not exist.")

        with zipfile.ZipFile(zip_file_name, "w", zipfile.ZIP_DEFLATED) as zipf:
            for root, dirs, files in os.walk(folder_path):
                dirs[:] = [d for d in dirs if d not in exclude_dirs]
                for file in files:
                    file_path = os.path.join(root, file)
                    arcname = os.path.relpath(file_path, folder_path)
                    zipf.write(file_path, arcname)

        print(f"Zipped {folder_path} into {zip_file_name}.")

    try:
        print(f"Zipping repository: {repository_dir}")
        zip_folder(repository_dir, artifact_path)

        # Random Unique ID - file_4charHex_HHMMSS
        artifact_id = f"artifacts_{uuid.uuid4().hex[:4]}_{datetime.datetime.now().strftime('%H%M%S')}.zip"
        key = artifact_prefix.format(artifact_id=artifact_id)

        print("Uploading to S3...")
        core.upload_generated_artifact(key, filename=artifact_path)

        print("Generating presigned URL...")
        artifact_url = core.get_generated_artifact_url(artifact_id)

        core.set_local_value("final_artifact_url", artifact_url)

        print(f"Generated Artifact URL: {artifact_url}")
        return artifact_url

    except Exception as e:
        return f"Error generating artifacts: {e}"

core.output(generate_artifacts_url(**core.args()))

Assemble the Agent

Now create a new Super Agent and combine all of the pieces we've made during this lab.

We also need to add a static prompt for the run_mesh_code tool. Do this by clicking on the tool after you've dragged it in.

* A static prompt is set in the run_mesh_code tool, so Claude always runs the same prompt for the agent, regardless of the codebase.

Now you can test it in either of two ways:

1. Test directly in Agent Manager:
   1. Navigate to the Uploaded Artifacts tab.
   2. Upload a zipped codebase.
   3. Give the file a name (preferably the same name as the file itself, with the file extension — e.g. abc.zip).
   4. Run the Agent and give the name of the file you uploaded as input.
2. Create a tool and persona in CUI, then upload the file in CUI and trigger.

You can use the following sample codebase for your testing:

And that's it!

This is basically the core flow that constitutes most of the Foundation Agents when it comes to reverse / forward engineering.

Meaning you could simply change the prompt for the run_mesh_code tool to do other things — for example:

• Security review and fix implementation
• Module/Library version upgrade code impact analysis
• UI/UX Enhancement

You can also chain run_mesh_code tool nodes sequentially with different prompts to form a pipeline — for example, Implement CR → Security Review → Generate Documentation.

Accuracy and expected output may be a different story though. If you had a 10-million-LOC, poorly written codebase, your mileage may obviously vary — but that barrier is decreasing by the day with newer and stronger LLM models.

You may also employ strategies like RAG, Knowledge Graph and existing documentation to improve results.

Github Integration

Publishing

Coming soon.

Glossary

Coming soon.