Agent

Contained within this file are experimental interfaces for working with the Synapse Python Client. Unless otherwise noted these interfaces are subject to change at any time. Use at your own risk.

Example Script:

Working with Synapse agents

"""
The purpose of this script is to demonstrate how to use the new OOP interface for Synapse AI Agents.

1. Register and send a prompt to a custom agent
2. Send a prompt to the baseline Synapse Agent
3. Conduct more than one session with the same agent
4. Start a new session with a custom agent and send a prompt to it
5. Start a new session with the baseline Synapse Agent and send a prompt to it
6. Start a new session with a custom agent and then update what the agent has access to
"""

import synapseclient
from synapseclient.models import Agent, AgentSession, AgentSessionAccessLevel

# IDs for a bedrock agent with the instructions:
# "You are a test agent that when greeted with: 'hello' will always response with: 'world'"
CLOUD_AGENT_ID = "QOTV3KQM1X"
AGENT_REGISTRATION_ID = 29

syn = synapseclient.Synapse(debug=True)
syn.login()

# Using the Agent class


# Register a custom agent and send a prompt to it
def register_and_send_prompt_to_custom_agent():
    my_custom_agent = Agent(cloud_agent_id=CLOUD_AGENT_ID)
    my_custom_agent.register(synapse_client=syn)
    my_custom_agent.prompt(
        prompt="Hello", enable_trace=True, print_response=True, synapse_client=syn
    )


# Create an Agent Object and prompt.
# By default, this will send a prompt to a new session with the baseline Synapse Agent.
def get_baseline_agent_and_send_prompt_to_it():
    baseline_agent = Agent()
    baseline_agent.prompt(
        prompt="What is Synapse?",
        enable_trace=True,
        print_response=True,
        synapse_client=syn,
    )


# Conduct more than one session with the same agent
def conduct_multiple_sessions_with_same_agent():
    my_agent = Agent(registration_id=AGENT_REGISTRATION_ID).get(synapse_client=syn)
    my_agent.prompt(
        prompt="Hello",
        enable_trace=True,
        print_response=True,
        synapse_client=syn,
    )
    my_second_session = my_agent.start_session(synapse_client=syn)
    my_agent.prompt(
        prompt="Hello again",
        enable_trace=True,
        print_response=True,
        session=my_second_session,
        synapse_client=syn,
    )


# Using the AgentSession class


# Start a new session with a custom agent and send a prompt to it
def start_new_session_with_custom_agent_and_send_prompt_to_it():
    my_session = AgentSession(agent_registration_id=AGENT_REGISTRATION_ID).start(
        synapse_client=syn
    )
    my_session.prompt(
        prompt="Hello", enable_trace=True, print_response=True, synapse_client=syn
    )


# Start a new session with the baseline Synapse Agent and send a prompt to it
def start_new_session_with_baseline_agent_and_send_prompt_to_it():
    my_session = AgentSession().start(synapse_client=syn)
    my_session.prompt(
        prompt="What is Synapse?",
        enable_trace=True,
        print_response=True,
        synapse_client=syn,
    )


# Start a new session with a custom agent and then update what the agent has access to
def start_new_session_with_custom_agent_and_update_access_to_it():
    my_session = AgentSession(agent_registration_id=AGENT_REGISTRATION_ID).start(
        synapse_client=syn
    )
    print(f"Access level before update: {my_session.access_level}")
    my_session.access_level = AgentSessionAccessLevel.READ_YOUR_PRIVATE_DATA
    my_session.update(synapse_client=syn)
    print(f"Access level after update: {my_session.access_level}")


register_and_send_prompt_to_custom_agent()
get_baseline_agent_and_send_prompt_to_it()
conduct_multiple_sessions_with_same_agent()
start_new_session_with_baseline_agent_and_send_prompt_to_it()
start_new_session_with_custom_agent_and_update_access_to_it()

API Reference

synapseclient.models.Agent dataclass

Bases: AgentSynchronousProtocol

Represents a Synapse Agent Registration

ATTRIBUTE	DESCRIPTION
cloud_agent_id	The unique ID of the agent in the cloud provider. TYPE: Optional[str]
cloud_alias_id	The alias ID of the agent in the cloud provider. Defaults to 'TSTALIASID' in the Synapse API. TYPE: Optional[str]
registration_id	The ID number of the agent assigned by Synapse. TYPE: Optional[int]
registered_on	The date the agent was registered. TYPE: Optional[datetime]
type	The type of agent. TYPE: Optional[AgentType]
sessions	A dictionary of AgentSession objects, keyed by session ID. TYPE: Dict[str, AgentSession]
current_session	The current session. Prompts will be sent to this session by default. TYPE: Optional[AgentSession]

Chat with the baseline Synapse Agent

You can chat with the same agent which is available in the Synapse UI at https://www.synapse.org/Chat:default. By default, this "baseline" agent is used when a registration ID is not provided. In the background, the Agent class will start a session and set that new session as the current session if one is not already set.

from synapseclient import Synapse
from synapseclient.models.agent import Agent

syn = Synapse()
syn.login()

my_agent = Agent()
my_agent.prompt(
    prompt="Can you tell me about the AD Knowledge Portal dataset?",
    enable_trace=True,
    print_response=True,
)

Register and chat with a custom agent

Only available for internal users (Sage Bionetworks employees)

Alternatively, you can register a custom agent and chat with it provided you have already created it.

from synapseclient import Synapse
from synapseclient.models.agent import Agent

syn = Synapse()
syn.login()

my_agent = Agent(cloud_agent_id="foo")
my_agent.register()
my_agent.prompt(
    prompt="Hello",
    enable_trace=True,
    print_response=True,
)

Get and chat with an existing agent

Retrieve an existing agent by providing the agent's registration ID and calling get(). Then, send a prompt to the agent.

from synapseclient import Synapse
from synapseclient.models.agent import Agent

syn = Synapse()
syn.login()

my_agent = Agent(registration_id="foo").get()
my_agent.prompt(
    prompt="Hello",
    enable_trace=True,
    print_response=True,
)

Start and prompt multiple sessions

Here, we connect to a custom agent and start one session with the prompt "Hello". In the background, this first session is being set as the current session and future prompts will be sent to this session by default. If we want to send a prompt to a different session, we can do so by starting it and calling prompt again, but with our new session as an argument. We now have two sessions, both stored in the my_agent.sessions dictionary. After the second prompt, my_second_session is now the current session.

syn = Synapse()
syn.login()

my_agent = Agent(registration_id="foo").get()

my_agent.prompt(
    prompt="Hello",
    enable_trace=True,
    print_response=True,
)

my_second_session = my_agent.start_session()
my_agent.prompt(
    prompt="Hello again",
    enable_trace=True,
    print_response=True,
    session=my_second_session,
)

Source code in synapseclient/models/agent.py

@dataclass
@async_to_sync
class Agent(AgentSynchronousProtocol):
    """Represents a [Synapse Agent Registration](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/AgentRegistration.html)

    Attributes:
        cloud_agent_id: The unique ID of the agent in the cloud provider.
        cloud_alias_id: The alias ID of the agent in the cloud provider.
                    Defaults to 'TSTALIASID' in the Synapse API.
        registration_id: The ID number of the agent assigned by Synapse.
        registered_on: The date the agent was registered.
        type: The type of agent.
        sessions: A dictionary of AgentSession objects, keyed by session ID.
        current_session: The current session. Prompts will be sent to this session by default.

    Example: Chat with the baseline Synapse Agent
        You can chat with the same agent which is available in the Synapse UI
        at https://www.synapse.org/Chat:default. By default, this "baseline" agent
        is used when a registration ID is not provided. In the background,
        the Agent class will start a session and set that new session as the
        current session if one is not already set.

            from synapseclient import Synapse
            from synapseclient.models.agent import Agent

            syn = Synapse()
            syn.login()

            my_agent = Agent()
            my_agent.prompt(
                prompt="Can you tell me about the AD Knowledge Portal dataset?",
                enable_trace=True,
                print_response=True,
            )

    Example: Register and chat with a custom agent
        **Only available for internal users (Sage Bionetworks employees)**

        Alternatively, you can register a custom agent and chat with it provided
        you have already created it.

            from synapseclient import Synapse
            from synapseclient.models.agent import Agent

            syn = Synapse()
            syn.login()

            my_agent = Agent(cloud_agent_id="foo")
            my_agent.register()
            my_agent.prompt(
                prompt="Hello",
                enable_trace=True,
                print_response=True,
            )

    Example: Get and chat with an existing agent
        Retrieve an existing agent by providing the agent's registration ID and calling `get()`.
        Then, send a prompt to the agent.

            from synapseclient import Synapse
            from synapseclient.models.agent import Agent

            syn = Synapse()
            syn.login()

            my_agent = Agent(registration_id="foo").get()
            my_agent.prompt(
                prompt="Hello",
                enable_trace=True,
                print_response=True,
            )

    Advanced Example: Start and prompt multiple sessions
        Here, we connect to a custom agent and start one session with the prompt "Hello".
        In the background, this first session is being set as the current session
        and future prompts will be sent to this session by default. If we want to send a
        prompt to a different session, we can do so by starting it and calling prompt again,
        but with our new session as an argument. We now have two sessions, both stored in the
        `my_agent.sessions` dictionary. After the second prompt, `my_second_session` is now
        the current session.

            syn = Synapse()
            syn.login()

            my_agent = Agent(registration_id="foo").get()

            my_agent.prompt(
                prompt="Hello",
                enable_trace=True,
                print_response=True,
            )

            my_second_session = my_agent.start_session()
            my_agent.prompt(
                prompt="Hello again",
                enable_trace=True,
                print_response=True,
                session=my_second_session,
            )
    """

    cloud_agent_id: Optional[str] = None
    """The unique ID of the agent in the cloud provider."""

    cloud_alias_id: Optional[str] = None
    """The alias ID of the agent in the cloud provider.
        Defaults to 'TSTALIASID' in the Synapse API.
    """

    registration_id: Optional[int] = None
    """The ID number of the agent assigned by Synapse."""

    registered_on: Optional[datetime] = None
    """The date the agent was registered."""

    type: Optional[AgentType] = None
    """The type of agent. One of either BASELINE or CUSTOM."""

    sessions: Dict[str, AgentSession] = field(default_factory=dict)
    """A dictionary of AgentSession objects, keyed by session ID."""

    current_session: Optional[AgentSession] = None
    """The current session. Prompts will be sent to this session by default."""

    def fill_from_dict(self, agent_registration: Dict[str, str]) -> "Agent":
        """
        Converts a response from the REST API into this dataclass.

        Arguments:
            agent_registration: The response from the REST API.

        Returns:
            The Agent object.
        """
        self.cloud_agent_id = agent_registration.get("awsAgentId", None)
        self.cloud_alias_id = agent_registration.get("awsAliasId", None)
        self.registration_id = agent_registration.get("agentRegistrationId", None)
        self.registered_on = agent_registration.get("registeredOn", None)
        self.type = (
            AgentType(agent_registration.get("type"))
            if agent_registration.get("type", None)
            else None
        )
        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Register_Agent: {self.registration_id}"
    )
    async def register_async(
        self, *, synapse_client: Optional[Synapse] = None
    ) -> "Agent":
        """Registers an agent with the Synapse API.
        If agent already exists, it will be retrieved.

        Arguments:
            synapse_client: If not passed in and caching was not disabled by
                    `Synapse.allow_client_caching(False)` this will use the last created
                    instance from the Synapse class constructor.

        Returns:
            The registered or existing Agent object.

        Example: Register and chat with a custom agent
            **Only available for internal users (Sage Bionetworks employees)**

            Alternatively, you can register a custom agent and chat with it provided
            you have already created it.

                import asyncio
                from synapseclient import Synapse
                from synapseclient.models.agent import Agent

                syn = Synapse()
                syn.login()

                async def main():
                    my_agent = Agent(cloud_agent_id="foo")
                    await my_agent.register_async()
                    await my_agent.prompt_async(
                        prompt="Hello",
                        enable_trace=True,
                        print_response=True,
                    )

                asyncio.run(main())
        """
        agent_response = await register_agent(
            cloud_agent_id=self.cloud_agent_id,
            cloud_alias_id=self.cloud_alias_id,
            synapse_client=synapse_client,
        )
        return self.fill_from_dict(agent_registration=agent_response)

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Get_Agent: {self.registration_id}"
    )
    async def get_async(self, *, synapse_client: Optional[Synapse] = None) -> "Agent":
        """Gets an existing custom agent. There is no need to use this method
        if you are trying to use the baseline agent.

        Arguments:
            synapse_client: If not passed in and caching was not disabled by
                    `Synapse.allow_client_caching(False)` this will use the last created
                    instance from the Synapse class constructor.

        Returns:
            The existing Agent object.

        Example: Get and chat with an existing agent
            Retrieve an existing custom agent by providing the agent's registration ID and calling `get_async()`.
            Then, send a prompt to the agent.

                import asyncio
                from synapseclient import Synapse
                from synapseclient.models import Agent, AgentSessionAccessLevel

                syn = Synapse()
                syn.login()

                async def main():
                    my_agent = await Agent(registration_id="foo").get_async()
                    await my_agent.prompt_async(
                        prompt="Hello",
                        enable_trace=True,
                        print_response=True,
                    )

                asyncio.run(main())
        """
        if self.registration_id is None:
            raise ValueError(
                "Registration ID is required to retrieve a custom agent. "
                "If you are trying to use the baseline agent, you do not need to "
                "use `get` or `get_async`. Instead, simply create an `Agent` object "
                "and start prompting `my_agent = Agent(); my_agent.prompt(...)`.",
            )
        agent_response = await get_agent(
            registration_id=self.registration_id,
            synapse_client=synapse_client,
        )
        return self.fill_from_dict(agent_registration=agent_response)

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Start_Agent_Session: {self.registration_id}"
    )
    async def start_session_async(
        self,
        access_level: Optional[
            AgentSessionAccessLevel
        ] = AgentSessionAccessLevel.PUBLICLY_ACCESSIBLE,
        *,
        synapse_client: Optional[Synapse] = None,
    ) -> "AgentSession":
        """Starts an agent session.
        Adds the session to the Agent's sessions dictionary and sets it as the current session.

        Arguments:
            access_level: The access level of the agent session.
                Must be one of PUBLICLY_ACCESSIBLE, READ_YOUR_PRIVATE_DATA,
                or WRITE_YOUR_PRIVATE_DATA.
                Defaults to PUBLICLY_ACCESSIBLE.
            synapse_client: If not passed in and caching was not disabled by
                    `Synapse.allow_client_caching(False)` this will use the last created
                    instance from the Synapse class constructor.

        Returns:
            The new AgentSession object.

        Example: Start a session and send a prompt with the baseline Synapse Agent.
            The baseline Synapse Agent is the default agent used when a registration ID is not provided.

                import asyncio
                from synapseclient import Synapse
                from synapseclient.models.agent import Agent

                syn = Synapse()
                syn.login()

                async def main():
                    my_agent = Agent()
                    await my_agent.start_session_async()
                    await my_agent.prompt_async(
                        prompt="Can you tell me about the AD Knowledge Portal dataset?",
                        enable_trace=True,
                        print_response=True,
                    )

                asyncio.run(main())

        Example: Start a session and send a prompt with a custom agent.
            The baseline Synapse Agent is the default agent used when a registration ID is not provided.

                import asyncio
                from synapseclient import Synapse
                from synapseclient.models.agent import Agent

                syn = Synapse()
                syn.login()

                async def main():
                    my_agent = Agent(cloud_agent_id="foo")
                    await my_agent.start_session_async()
                    await my_agent.prompt_async(
                        prompt="Hello",
                        enable_trace=True,
                        print_response=True,
                    )

                asyncio.run(main())
        """
        access_level = AgentSessionAccessLevel(access_level)
        session = await AgentSession(
            agent_registration_id=self.registration_id, access_level=access_level
        ).start_async(synapse_client=synapse_client)
        self.sessions[session.id] = session
        self.current_session = session
        return session

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Get_Agent_Session: {self.registration_id}"
    )
    async def get_session_async(
        self, session_id: str, *, synapse_client: Optional[Synapse] = None
    ) -> "AgentSession":
        """Gets an existing agent session.
        Adds the session to the Agent's sessions dictionary and
        sets it as the current session.

        Arguments:
            session_id: The ID of the session to get.
            synapse_client: If not passed in and caching was not disabled by
                    `Synapse.allow_client_caching(False)` this will use the last created
                    instance from the Synapse class constructor.

        Returns:
            The existing AgentSession object.

        Example: Get an existing session and send a prompt.
            Retrieve an existing session by providing the session ID and calling `get()`.
            Then, send a prompt to the agent.

                import asyncio
                from synapseclient import Synapse
                from synapseclient.models.agent import Agent

                syn = Synapse()
                syn.login()

                async def main():
                    my_session = await Agent().get_session_async(session_id="foo")
                    await my_session.prompt_async(
                        prompt="Hello",
                        enable_trace=True,
                        print_response=True,
                    )

                asyncio.run(main())
        """
        session = await AgentSession(id=session_id).get_async(
            synapse_client=synapse_client
        )
        if session.id not in self.sessions:
            self.sessions[session.id] = session
        self.current_session = session
        return session

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Prompt_Agent_Session: {self.registration_id}"
    )
    async def prompt_async(
        self,
        prompt: str,
        enable_trace: bool = False,
        print_response: bool = False,
        session: Optional[AgentSession] = None,
        newer_than: Optional[int] = None,
        *,
        synapse_client: Optional[Synapse] = None,
    ) -> AgentPrompt:
        """Sends a prompt to the agent for the current session.
            If no session is currently active, a new session will be started.

        Arguments:
            prompt: The prompt to send to the agent.
            enable_trace: Whether to enable trace for the prompt.
            print_response: Whether to print the response to the console.
            session_id: The ID of the session to send the prompt to.
                If None, the current session will be used.
            newer_than: The timestamp to get trace results newer than. Defaults to None (all results).
            synapse_client: If not passed in and caching was not disabled by
                    `Synapse.allow_client_caching(False)` this will use the last created
                    instance from the Synapse class constructor.

        Example: Prompt the baseline Synapse Agent.
            The baseline Synapse Agent is equivilent to the Agent available in the Synapse UI.

                import asyncio
                from synapseclient import Synapse
                from synapseclient.models.agent import Agent

                syn = Synapse()
                syn.login()

                async def main():
                    my_agent = Agent()
                    await my_agent.prompt_async(
                        prompt="Can you tell me about the AD Knowledge Portal dataset?",
                        enable_trace=True,
                        print_response=True,
                    )

                asyncio.run(main())

        Example: Prompt a custom agent.
            If you have already registered a custom agent, you can prompt it by providing the agent's registration ID.

                import asyncio
                from synapseclient import Synapse
                from synapseclient.models.agent import Agent

                syn = Synapse()
                syn.login()

                async def main():
                    my_agent = Agent(registration_id="foo")
                    await my_agent.prompt_async(
                        prompt="Hello",
                        enable_trace=True,
                        print_response=True,
                    )

                asyncio.run(main())

        Advanced Example: Start and prompt multiple sessions
            Here, we connect to a custom agent and start one session with the prompt "Hello".
            In the background, this first session is being set as the current session
            and future prompts will be sent to this session by default. If we want to send a
            prompt to a different session, we can do so by starting it and calling prompt again,
            but with our new session as an argument. We now have two sessions, both stored in the
            `my_agent.sessions` dictionary. After the second prompt, `my_second_session` is now
            the current session.

                import asyncio
                from synapseclient import Synapse
                from synapseclient.models.agent import Agent

                syn = Synapse()
                syn.login()

                async def main():
                    my_agent = Agent(registration_id="foo").get()
                    await my_agent.prompt_async(
                        prompt="Hello",
                        enable_trace=True,
                        print_response=True,
                    )

                    my_second_session = await my_agent.start_session_async()
                    await my_agent.prompt_async(
                        prompt="Hello again",
                        enable_trace=True,
                        print_response=True,
                        session=my_second_session,
                    )

                asyncio.run(main())
        """
        if session:
            await self.get_session_async(
                session_id=session.id, synapse_client=synapse_client
            )
        else:
            if not self.current_session:
                await self.start_session_async(synapse_client=synapse_client)

        return await self.current_session.prompt_async(
            prompt=prompt,
            enable_trace=enable_trace,
            newer_than=newer_than,
            print_response=print_response,
            synapse_client=synapse_client,
        )

    def get_chat_history(self) -> Union[List[AgentPrompt], None]:
        """Gets the chat history for the current session.

        Example: Get the chat history for the current session.
            First, send a prompt to the agent.
            Then, retrieve the chat history for the current session by calling `get_chat_history()`.

                import asyncio
                from synapseclient import Synapse
                from synapseclient.models.agent import Agent

                syn = Synapse()
                syn.login()

                async def main():
                    my_agent = Agent()
                    await my_agent.prompt_async(
                        prompt="Hello",
                        enable_trace=True,
                        print_response=True,
                    )
                    print(my_agent.get_chat_history())

                asyncio.run(main())
        """
        return self.current_session.chat_history if self.current_session else None

Functions

register

register(*, synapse_client: Optional[Synapse] = None) -> Agent

Registers an agent with the Synapse API. If agent already exists, it will be retrieved.

PARAMETER	DESCRIPTION
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Agent	The registered or existing Agent object.

Register and chat with a custom agent

Only available for internal users (Sage Bionetworks employees)

Alternatively, you can register a custom agent and chat with it provided you have already created it.

from synapseclient import Synapse
from synapseclient.models.agent import Agent

syn = Synapse()
syn.login()

my_agent = Agent(cloud_agent_id="foo")
my_agent.register()

my_agent.prompt(
    prompt="Hello",
    enable_trace=True,
    print_response=True,
)

Source code in synapseclient/models/protocols/agent_protocol.py

def register(self, *, synapse_client: Optional[Synapse] = None) -> "Agent":
    """Registers an agent with the Synapse API.
    If agent already exists, it will be retrieved.

    Arguments:
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The registered or existing Agent object.

    Example: Register and chat with a custom agent
        **Only available for internal users (Sage Bionetworks employees)**

        Alternatively, you can register a custom agent and chat with it provided
        you have already created it.

            from synapseclient import Synapse
            from synapseclient.models.agent import Agent

            syn = Synapse()
            syn.login()

            my_agent = Agent(cloud_agent_id="foo")
            my_agent.register()

            my_agent.prompt(
                prompt="Hello",
                enable_trace=True,
                print_response=True,
            )
    """
    return self

get

get(*, synapse_client: Optional[Synapse] = None) -> Agent

Gets an existing agent.

PARAMETER	DESCRIPTION
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Agent	The existing Agent object.

Get and chat with an existing agent

Retrieve an existing agent by providing the agent's registration ID and calling get(). Then, send a prompt to the agent.

from synapseclient import Synapse
from synapseclient.models.agent import Agent

syn = Synapse()
syn.login()

my_agent = Agent(registration_id="foo").get()
my_agent.prompt(
    prompt="Hello",
    enable_trace=True,
    print_response=True,
)

Source code in synapseclient/models/protocols/agent_protocol.py

def get(self, *, synapse_client: Optional[Synapse] = None) -> "Agent":
    """Gets an existing agent.

    Arguments:
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The existing Agent object.

    Example: Get and chat with an existing agent
        Retrieve an existing agent by providing the agent's registration ID and calling `get()`.
        Then, send a prompt to the agent.

            from synapseclient import Synapse
            from synapseclient.models.agent import Agent

            syn = Synapse()
            syn.login()

            my_agent = Agent(registration_id="foo").get()
            my_agent.prompt(
                prompt="Hello",
                enable_trace=True,
                print_response=True,
            )
    """
    return self

start_session

start_session(access_level: Optional[AgentSessionAccessLevel] = 'PUBLICLY_ACCESSIBLE', *, synapse_client: Optional[Synapse] = None) -> AgentSession

Starts an agent session. Adds the session to the Agent's sessions dictionary and sets it as the current session.

PARAMETER	DESCRIPTION
access_level	The access level of the agent session. Must be one of PUBLICLY_ACCESSIBLE, READ_YOUR_PRIVATE_DATA, or WRITE_YOUR_PRIVATE_DATA. Defaults to PUBLICLY_ACCESSIBLE. TYPE: Optional[AgentSessionAccessLevel] DEFAULT: 'PUBLICLY_ACCESSIBLE'
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
AgentSession	The new AgentSession object.

Start a session and send a prompt with the baseline Synapse Agent.

The baseline Synapse Agent is the default agent used when a registration ID is not provided.

from synapseclient import Synapse
from synapseclient.models.agent import Agent

syn = Synapse()
syn.login()

my_agent = Agent()
my_agent.start_session()
my_agent.prompt(
    prompt="Can you tell me about the AD Knowledge Portal dataset?",
    enable_trace=True,
    print_response=True,
)

Start a session and send a prompt with a custom agent.

The baseline Synapse Agent is the default agent used when a registration ID is not provided.

from synapseclient import Synapse
from synapseclient.models.agent import Agent

syn = Synapse()
syn.login()

my_agent = Agent(cloud_agent_id="foo")
my_agent.start_session()
my_agent.prompt(
    prompt="Hello",
    enable_trace=True,
    print_response=True,
)

Source code in synapseclient/models/protocols/agent_protocol.py

def start_session(
    self,
    access_level: Optional["AgentSessionAccessLevel"] = "PUBLICLY_ACCESSIBLE",
    *,
    synapse_client: Optional[Synapse] = None,
) -> "AgentSession":
    """Starts an agent session.
    Adds the session to the Agent's sessions dictionary and sets it as the current session.

    Arguments:
        access_level: The access level of the agent session.
            Must be one of PUBLICLY_ACCESSIBLE, READ_YOUR_PRIVATE_DATA,
            or WRITE_YOUR_PRIVATE_DATA.
            Defaults to PUBLICLY_ACCESSIBLE.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The new AgentSession object.

    Example: Start a session and send a prompt with the baseline Synapse Agent.
        The baseline Synapse Agent is the default agent used when a registration ID is not provided.

            from synapseclient import Synapse
            from synapseclient.models.agent import Agent

            syn = Synapse()
            syn.login()

            my_agent = Agent()
            my_agent.start_session()
            my_agent.prompt(
                prompt="Can you tell me about the AD Knowledge Portal dataset?",
                enable_trace=True,
                print_response=True,
            )

    Example: Start a session and send a prompt with a custom agent.
        The baseline Synapse Agent is the default agent used when a registration ID is not provided.

            from synapseclient import Synapse
            from synapseclient.models.agent import Agent

            syn = Synapse()
            syn.login()

            my_agent = Agent(cloud_agent_id="foo")
            my_agent.start_session()
            my_agent.prompt(
                prompt="Hello",
                enable_trace=True,
                print_response=True,
            )
    """
    return AgentSession()

get_session

get_session(session_id: str, *, synapse_client: Optional[Synapse] = None) -> AgentSession

Gets an existing agent session. Adds the session to the Agent's sessions dictionary and sets it as the current session.

PARAMETER	DESCRIPTION
session_id	The ID of the session to get. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
AgentSession	The existing AgentSession object.

Get an existing session and send a prompt.

Retrieve an existing session by providing the session ID and calling get(). Then, send a prompt to the agent.

from synapseclient import Synapse
from synapseclient.models.agent import Agent

syn = Synapse()
syn.login()

my_session = Agent().get_session(session_id="foo")
my_session.prompt(
    prompt="Hello",
    enable_trace=True,
    print_response=True,
)

Source code in synapseclient/models/protocols/agent_protocol.py

def get_session(
    self, session_id: str, *, synapse_client: Optional[Synapse] = None
) -> "AgentSession":
    """Gets an existing agent session.
    Adds the session to the Agent's sessions dictionary and
    sets it as the current session.

    Arguments:
        session_id: The ID of the session to get.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The existing AgentSession object.

    Example: Get an existing session and send a prompt.
        Retrieve an existing session by providing the session ID and calling `get()`.
        Then, send a prompt to the agent.

            from synapseclient import Synapse
            from synapseclient.models.agent import Agent

            syn = Synapse()
            syn.login()

            my_session = Agent().get_session(session_id="foo")
            my_session.prompt(
                prompt="Hello",
                enable_trace=True,
                print_response=True,
            )
    """
    return AgentSession()

prompt

prompt(prompt: str, enable_trace: bool = False, print_response: bool = False, session: Optional[AgentSession] = None, newer_than: Optional[int] = None, *, synapse_client: Optional[Synapse] = None) -> AgentPrompt

Sends a prompt to the agent for the current session. If no session is currently active, a new session will be started.

PARAMETER	DESCRIPTION
prompt	The prompt to send to the agent. TYPE: str
enable_trace	Whether to enable trace for the prompt. TYPE: bool DEFAULT: False
print_response	Whether to print the response to the console. TYPE: bool DEFAULT: False
session_id	The ID of the session to send the prompt to. If None, the current session will be used.
newer_than	The timestamp to get trace results newer than. Defaults to None (all results). TYPE: Optional[int] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Prompt the baseline Synapse Agent.

The baseline Synapse Agent is equivilent to the Agent available in the Synapse UI.

from synapseclient import Synapse from synapseclient.models import Agent

syn = Synapse() syn.login()

my_agent = Agent() my_agent.prompt( prompt="Can you tell me about the AD Knowledge Portal dataset?", enable_trace=True, print_response=True, )

Prompt a custom agent.

If you have already registered a custom agent, you can prompt it by providing the agent's registration ID.

from synapseclient import Synapse
from synapseclient.models.agent import Agent

syn = Synapse()
syn.login()

my_agent = Agent(registration_id="foo")
my_agent.prompt(
    prompt="Hello",
    enable_trace=True,
    print_response=True,
)

Start and prompt multiple sessions

Here, we connect to a custom agent and start one session with the prompt "Hello". In the background, this first session is being set as the current session and future prompts will be sent to this session by default. If we want to send a prompt to a different session, we can do so by starting it and calling prompt again, but with our new session as an argument. We now have two sessions, both stored in the my_agent.sessions dictionary. After the second prompt, my_second_session is now the current session.

syn = Synapse()
syn.login()

my_agent = Agent(registration_id="foo").get()

my_agent.prompt(
    prompt="Hello",
    enable_trace=True,
    print_response=True,
)

my_second_session = my_agent.start_session()
my_agent.prompt(
    prompt="Hello again",
    enable_trace=True,
    print_response=True,
    session=my_second_session,
)

Source code in synapseclient/models/protocols/agent_protocol.py

def prompt(
    self,
    prompt: str,
    enable_trace: bool = False,
    print_response: bool = False,
    session: Optional["AgentSession"] = None,
    newer_than: Optional[int] = None,
    *,
    synapse_client: Optional[Synapse] = None,
) -> "AgentPrompt":
    """Sends a prompt to the agent for the current session.
        If no session is currently active, a new session will be started.

    Arguments:
        prompt: The prompt to send to the agent.
        enable_trace: Whether to enable trace for the prompt.
        print_response: Whether to print the response to the console.
        session_id: The ID of the session to send the prompt to.
            If None, the current session will be used.
        newer_than: The timestamp to get trace results newer than. Defaults to None (all results).
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Example: Prompt the baseline Synapse Agent.
        The baseline Synapse Agent is equivilent to the Agent available in the Synapse UI.

        from synapseclient import Synapse
        from synapseclient.models import Agent

        syn = Synapse()
        syn.login()

        my_agent = Agent()
        my_agent.prompt(
            prompt="Can you tell me about the AD Knowledge Portal dataset?",
            enable_trace=True,
            print_response=True,
        )

    Example: Prompt a custom agent.
        If you have already registered a custom agent, you can prompt it by providing the agent's registration ID.

            from synapseclient import Synapse
            from synapseclient.models.agent import Agent

            syn = Synapse()
            syn.login()

            my_agent = Agent(registration_id="foo")
            my_agent.prompt(
                prompt="Hello",
                enable_trace=True,
                print_response=True,
            )

    Advanced Example: Start and prompt multiple sessions
        Here, we connect to a custom agent and start one session with the prompt "Hello".
        In the background, this first session is being set as the current session
        and future prompts will be sent to this session by default. If we want to send a
        prompt to a different session, we can do so by starting it and calling prompt again,
        but with our new session as an argument. We now have two sessions, both stored in the
        `my_agent.sessions` dictionary. After the second prompt, `my_second_session` is now
        the current session.

            syn = Synapse()
            syn.login()

            my_agent = Agent(registration_id="foo").get()

            my_agent.prompt(
                prompt="Hello",
                enable_trace=True,
                print_response=True,
            )

            my_second_session = my_agent.start_session()
            my_agent.prompt(
                prompt="Hello again",
                enable_trace=True,
                print_response=True,
                session=my_second_session,
            )
    """
    return AgentPrompt()

get_chat_history

get_chat_history() -> Union[List[AgentPrompt], None]

Gets the chat history for the current session.

Get the chat history for the current session.

First, send a prompt to the agent. Then, retrieve the chat history for the current session by calling get_chat_history().

import asyncio
from synapseclient import Synapse
from synapseclient.models.agent import Agent

syn = Synapse()
syn.login()

async def main():
    my_agent = Agent()
    await my_agent.prompt_async(
        prompt="Hello",
        enable_trace=True,
        print_response=True,
    )
    print(my_agent.get_chat_history())

asyncio.run(main())

Source code in synapseclient/models/agent.py

def get_chat_history(self) -> Union[List[AgentPrompt], None]:
    """Gets the chat history for the current session.

    Example: Get the chat history for the current session.
        First, send a prompt to the agent.
        Then, retrieve the chat history for the current session by calling `get_chat_history()`.

            import asyncio
            from synapseclient import Synapse
            from synapseclient.models.agent import Agent

            syn = Synapse()
            syn.login()

            async def main():
                my_agent = Agent()
                await my_agent.prompt_async(
                    prompt="Hello",
                    enable_trace=True,
                    print_response=True,
                )
                print(my_agent.get_chat_history())

            asyncio.run(main())
    """
    return self.current_session.chat_history if self.current_session else None

---

synapseclient.models.AgentSession dataclass

Bases: AgentSessionSynchronousProtocol

Represents a Synapse Agent Session

ATTRIBUTE	DESCRIPTION
id	The unique ID of the agent session. Can only be used by the user that created it. TYPE: Optional[str]
access_level	The access level of the agent session. One of PUBLICLY_ACCESSIBLE, READ_YOUR_PRIVATE_DATA, or WRITE_YOUR_PRIVATE_DATA. TYPE: Optional[AgentSessionAccessLevel]
started_on	The date the agent session was started. TYPE: Optional[datetime]
started_by	The ID of the user who started the agent session. TYPE: Optional[int]
modified_on	The date the agent session was last modified. TYPE: Optional[datetime]
agent_registration_id	The registration ID of the agent that will be used for this session. TYPE: Optional[int]
etag	The etag of the agent session. TYPE: Optional[str]

Note: It is recommended to use the Agent class to conduct chat sessions, but you are free to use AgentSession directly if you wish.

Start a session and send a prompt.

Start a session with a custom agent by providing the agent's registration ID and calling start(). Then, send a prompt to the agent.

from synapseclient import Synapse
from synapseclient.models.agent import AgentSession

syn = Synapse()
syn.login()

my_session = AgentSession(agent_registration_id="foo").start()
my_session.prompt(
    prompt="Hello",
    enable_trace=True,
    print_response=True,
)

Get an existing session and send a prompt.

Retrieve an existing session by providing the session ID and calling get(). Then, send a prompt to the agent.

from synapseclient import Synapse
from synapseclient.models.agent import AgentSession

syn = Synapse()
syn.login()

my_session = AgentSession(id="foo").get()
my_session.prompt(
    prompt="Hello",
    enable_trace=True,
    print_response=True,
)

Update the access level of an existing session.

Retrieve an existing session by providing the session ID and calling get(). Then, update the access level of the session and call update().

from synapseclient import Synapse
from synapseclient.models.agent import AgentSession, AgentSessionAccessLevel

syn = Synapse()
syn.login()

my_session = AgentSession(id="foo").get()
my_session.access_level = AgentSessionAccessLevel.READ_YOUR_PRIVATE_DATA
my_session.update()

Source code in synapseclient/models/agent.py

@dataclass
@async_to_sync
class AgentSession(AgentSessionSynchronousProtocol):
    """Represents a [Synapse Agent Session](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/AgentSession.html)

    Attributes:
        id: The unique ID of the agent session.
            Can only be used by the user that created it.
        access_level: The access level of the agent session.
            One of PUBLICLY_ACCESSIBLE, READ_YOUR_PRIVATE_DATA,
            or WRITE_YOUR_PRIVATE_DATA.
        started_on: The date the agent session was started.
        started_by: The ID of the user who started the agent session.
        modified_on: The date the agent session was last modified.
        agent_registration_id: The registration ID of the agent that will
            be used for this session.
        etag: The etag of the agent session.

    Note: It is recommended to use the `Agent` class to conduct chat sessions,
    but you are free to use AgentSession directly if you wish.

    Example: Start a session and send a prompt.
        Start a session with a custom agent by providing the agent's registration ID and calling `start()`.
        Then, send a prompt to the agent.

            from synapseclient import Synapse
            from synapseclient.models.agent import AgentSession

            syn = Synapse()
            syn.login()

            my_session = AgentSession(agent_registration_id="foo").start()
            my_session.prompt(
                prompt="Hello",
                enable_trace=True,
                print_response=True,
            )

    Example: Get an existing session and send a prompt.
        Retrieve an existing session by providing the session ID and calling `get()`.
        Then, send a prompt to the agent.

            from synapseclient import Synapse
            from synapseclient.models.agent import AgentSession

            syn = Synapse()
            syn.login()

            my_session = AgentSession(id="foo").get()
            my_session.prompt(
                prompt="Hello",
                enable_trace=True,
                print_response=True,
            )

    Example: Update the access level of an existing session.
        Retrieve an existing session by providing the session ID and calling `get()`.
        Then, update the access level of the session and call `update()`.

            from synapseclient import Synapse
            from synapseclient.models.agent import AgentSession, AgentSessionAccessLevel

            syn = Synapse()
            syn.login()

            my_session = AgentSession(id="foo").get()
            my_session.access_level = AgentSessionAccessLevel.READ_YOUR_PRIVATE_DATA
            my_session.update()
    """

    id: Optional[str] = None
    """The unique ID of the agent session.
    Can only be used by the user that created it."""

    access_level: Optional[
        AgentSessionAccessLevel
    ] = AgentSessionAccessLevel.PUBLICLY_ACCESSIBLE
    """The access level of the agent session.
        One of PUBLICLY_ACCESSIBLE, READ_YOUR_PRIVATE_DATA, or
        WRITE_YOUR_PRIVATE_DATA. Defaults to PUBLICLY_ACCESSIBLE.
    """

    started_on: Optional[datetime] = None
    """The date the agent session was started."""

    started_by: Optional[int] = None
    """The ID of the user who started the agent session."""

    modified_on: Optional[datetime] = None
    """The date the agent session was last modified."""

    agent_registration_id: Optional[int] = None
    """The registration ID of the agent that will be used for this session."""

    etag: Optional[str] = None
    """The etag of the agent session."""

    chat_history: List[AgentPrompt] = field(default_factory=list)
    """A list of AgentPrompt objects."""

    def fill_from_dict(self, synapse_agent_session: Dict[str, str]) -> "AgentSession":
        """
        Converts a response from the REST API into this dataclass.

        Arguments:
            synapse_agent_session: The response from the REST API.

        Returns:
            The AgentSession object.
        """
        self.id = synapse_agent_session.get("sessionId", None)
        self.access_level = synapse_agent_session.get("agentAccessLevel", None)
        self.started_on = synapse_agent_session.get("startedOn", None)
        self.started_by = synapse_agent_session.get("startedBy", None)
        self.modified_on = synapse_agent_session.get("modifiedOn", None)
        self.agent_registration_id = synapse_agent_session.get(
            "agentRegistrationId", None
        )
        self.etag = synapse_agent_session.get("etag", None)
        return self

    @otel_trace_method(method_to_trace_name=lambda self, **kwargs: "Start_Session")
    async def start_async(
        self, *, synapse_client: Optional[Synapse] = None
    ) -> "AgentSession":
        """Starts an agent session.

        Arguments:
            synapse_client: If not passed in and caching was not disabled by
                    `Synapse.allow_client_caching(False)` this will use the last created
                    instance from the Synapse class constructor.

        Returns:
            The new AgentSession object.

        Example: Start a session and send a prompt.
            Start a session with a custom agent by providing the agent's registration ID and calling `start()`.
            Then, send a prompt to the agent.

                import asyncio
                from synapseclient import Synapse
                from synapseclient.models.agent import AgentSession

                syn = Synapse()
                syn.login()

                async def main():
                    my_session = await AgentSession(agent_registration_id="foo").start_async()
                    await my_session.prompt_async(
                        prompt="Hello",
                        enable_trace=True,
                        print_response=True,
                    )

                asyncio.run(main())
        """
        session_response = await start_session(
            access_level=self.access_level,
            agent_registration_id=self.agent_registration_id,
            synapse_client=synapse_client,
        )
        return self.fill_from_dict(synapse_agent_session=session_response)

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Get_Session: {self.id}"
    )
    async def get_async(
        self, *, synapse_client: Optional[Synapse] = None
    ) -> "AgentSession":
        """Gets an agent session.

        Arguments:
            synapse_client: If not passed in and caching was not disabled by
                    `Synapse.allow_client_caching(False)` this will use the last created
                    instance from the Synapse class constructor.

        Returns:
            The retrieved AgentSession object.

        Example: Get an existing session and send a prompt.
            Retrieve an existing session by providing the session ID and calling `get()`.
            Then, send a prompt to the agent.

                import asyncio
                from synapseclient import Synapse
                from synapseclient.models.agent import AgentSession

                syn = Synapse()
                syn.login()

                async def main():
                    my_session = await AgentSession(id="foo").get_async()
                    await my_session.prompt_async(
                        prompt="Hello",
                        enable_trace=True,
                        print_response=True,
                    )

                asyncio.run(main())
        """
        session_response = await get_session(
            id=self.id,
            synapse_client=synapse_client,
        )
        return self.fill_from_dict(synapse_agent_session=session_response)

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Update_Session: {self.id}"
    )
    async def update_async(
        self,
        *,
        synapse_client: Optional[Synapse] = None,
    ) -> "AgentSession":
        """Updates an agent session.
        Only updates to the access level are currently supported.

        Arguments:
            synapse_client: If not passed in and caching was not disabled by
                    `Synapse.allow_client_caching(False)` this will use the last created
                    instance from the Synapse class constructor.

        Returns:
            The updated AgentSession object.

        Example: Update the access level of an existing session.
            Retrieve an existing session by providing the session ID and calling `get()`.
            Then, update the access level of the session and call `update()`.

                import asyncio
                from synapseclient import Synapse
                from synapseclient.models.agent import AgentSession, AgentSessionAccessLevel

                syn = Synapse()
                syn.login()

                async def main():
                    my_session = await AgentSession(id="foo").get_async()
                    my_session.access_level = AgentSessionAccessLevel.READ_YOUR_PRIVATE_DATA
                    await my_session.update_async()

                asyncio.run(main())
        """
        session_response = await update_session(
            id=self.id,
            access_level=self.access_level,
            synapse_client=synapse_client,
        )
        return self.fill_from_dict(synapse_agent_session=session_response)

    @otel_trace_method(method_to_trace_name=lambda self, **kwargs: f"Prompt: {self.id}")
    async def prompt_async(
        self,
        prompt: str,
        enable_trace: bool = False,
        print_response: bool = False,
        newer_than: Optional[int] = None,
        *,
        synapse_client: Optional[Synapse] = None,
    ) -> AgentPrompt:
        """Sends a prompt to the agent and adds the response to the AgentSession's
        chat history. A session must be started before sending a prompt.

        Arguments:
            prompt: The prompt to send to the agent.
            enable_trace: Whether to enable trace for the prompt.
            print_response: Whether to print the response to the console.
            newer_than: The timestamp to get trace results newer than.
                Defaults to None (all results).
            synapse_client: If not passed in and caching was not disabled by
                    `Synapse.allow_client_caching(False)` this will use the last created
                    instance from the Synapse class constructor.

        Example: Send a prompt within an existing session.
            Retrieve an existing session by providing the session ID and calling `get()`.
            Then, send a prompt to the agent.

                import asyncio
                from synapseclient import Synapse
                from synapseclient.models.agent import AgentSession

                syn = Synapse()
                syn.login()

                async def main():
                    my_session = await AgentSession(id="foo").get_async()
                    await my_session.prompt_async(
                        prompt="Hello",
                        enable_trace=True,
                        print_response=True,
                    )

                asyncio.run(main())
        """
        agent_prompt = await AgentPrompt(
            prompt=prompt, session_id=self.id, enable_trace=enable_trace
        ).send_job_and_wait_async(
            synapse_client=synapse_client, post_exchange_args={"newer_than": newer_than}
        )
        self.chat_history.append(agent_prompt)
        if print_response:
            client = Synapse.get_client(synapse_client=synapse_client)
            client.logger.info(f"PROMPT:\n{prompt}\n")
            client.logger.info(f"RESPONSE:\n{agent_prompt.response}\n")
            if enable_trace:
                client.logger.info(f"TRACE:\n{agent_prompt.trace}")
        return agent_prompt

Functions

start

start(*, synapse_client: Optional[Synapse] = None) -> AgentSession

Starts an agent session.

PARAMETER	DESCRIPTION
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
AgentSession	The new AgentSession object.

Start a session and send a prompt.

Start a session with a custom agent by providing the agent's registration ID and calling start(). Then, send a prompt to the agent.

from synapseclient import Synapse
from synapseclient.models.agent import AgentSession

syn = Synapse()
syn.login()

my_session = AgentSession(agent_registration_id="foo").start()
my_session.prompt(
    prompt="Hello",
    enable_trace=True,
    print_response=True,
)

Source code in synapseclient/models/protocols/agent_protocol.py

def start(self, *, synapse_client: Optional[Synapse] = None) -> "AgentSession":
    """Starts an agent session.

    Arguments:
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The new AgentSession object.

    Example: Start a session and send a prompt.
        Start a session with a custom agent by providing the agent's registration ID and calling `start()`.
        Then, send a prompt to the agent.

            from synapseclient import Synapse
            from synapseclient.models.agent import AgentSession

            syn = Synapse()
            syn.login()

            my_session = AgentSession(agent_registration_id="foo").start()
            my_session.prompt(
                prompt="Hello",
                enable_trace=True,
                print_response=True,
            )
    """
    return self

get

get(*, synapse_client: Optional[Synapse] = None) -> AgentSession

Gets an agent session.

PARAMETER	DESCRIPTION
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
AgentSession	The retrieved AgentSession object.

Get an existing session and send a prompt.

Retrieve an existing session by providing the session ID and calling get(). Then, send a prompt to the agent.

from synapseclient import Synapse
from synapseclient.models.agent import AgentSession

syn = Synapse()
syn.login()

my_session = AgentSession(id="foo").get()
my_session.prompt(
    prompt="Hello",
    enable_trace=True,
    print_response=True,
)

Source code in synapseclient/models/protocols/agent_protocol.py

def get(self, *, synapse_client: Optional[Synapse] = None) -> "AgentSession":
    """Gets an agent session.

    Arguments:
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The retrieved AgentSession object.

    Example: Get an existing session and send a prompt.
        Retrieve an existing session by providing the session ID and calling `get()`.
        Then, send a prompt to the agent.

            from synapseclient import Synapse
            from synapseclient.models.agent import AgentSession

            syn = Synapse()
            syn.login()

            my_session = AgentSession(id="foo").get()
            my_session.prompt(
                prompt="Hello",
                enable_trace=True,
                print_response=True,
            )
    """
    return self

update

update(*, synapse_client: Optional[Synapse] = None) -> AgentSession

Updates an agent session. Only updates to the access level are currently supported.

PARAMETER	DESCRIPTION
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
AgentSession	The updated AgentSession object.

Update the access level of an existing session.

Retrieve an existing session by providing the session ID and calling get(). Then, update the access level of the session and call update().

from synapseclient import Synapse
from synapseclient.models.agent import AgentSession, AgentSessionAccessLevel

syn = Synapse()
syn.login()

my_session = AgentSession(id="foo").get()
my_session.access_level = AgentSessionAccessLevel.READ_YOUR_PRIVATE_DATA
my_session.update()

Source code in synapseclient/models/protocols/agent_protocol.py

def update(self, *, synapse_client: Optional[Synapse] = None) -> "AgentSession":
    """Updates an agent session.
    Only updates to the access level are currently supported.

    Arguments:
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The updated AgentSession object.

    Example: Update the access level of an existing session.
        Retrieve an existing session by providing the session ID and calling `get()`.
        Then, update the access level of the session and call `update()`.

            from synapseclient import Synapse
            from synapseclient.models.agent import AgentSession, AgentSessionAccessLevel

            syn = Synapse()
            syn.login()

            my_session = AgentSession(id="foo").get()
            my_session.access_level = AgentSessionAccessLevel.READ_YOUR_PRIVATE_DATA
            my_session.update()
    """
    return self

prompt

prompt(prompt: str, enable_trace: bool = False, print_response: bool = False, newer_than: Optional[int] = None, *, synapse_client: Optional[Synapse] = None) -> AgentPrompt

Sends a prompt to the agent and adds the response to the AgentSession's chat history. A session must be started before sending a prompt.

PARAMETER	DESCRIPTION
prompt	The prompt to send to the agent. TYPE: str
enable_trace	Whether to enable trace for the prompt. TYPE: bool DEFAULT: False
print_response	Whether to print the response to the console. TYPE: bool DEFAULT: False
newer_than	The timestamp to get trace results newer than. Defaults to None (all results). TYPE: Optional[int] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Send a prompt within an existing session.

Retrieve an existing session by providing the session ID and calling get(). Then, send a prompt to the agent.

from synapseclient import Synapse
from synapseclient.models.agent import AgentSession

syn = Synapse()
syn.login()

my_session = AgentSession(id="foo").get()
my_session.prompt(
    prompt="Hello",
    enable_trace=True,
    print_response=True,
)

Source code in synapseclient/models/protocols/agent_protocol.py

def prompt(
    self,
    prompt: str,
    enable_trace: bool = False,
    print_response: bool = False,
    newer_than: Optional[int] = None,
    *,
    synapse_client: Optional[Synapse] = None,
) -> "AgentPrompt":
    """Sends a prompt to the agent and adds the response to the AgentSession's
    chat history. A session must be started before sending a prompt.

    Arguments:
        prompt: The prompt to send to the agent.
        enable_trace: Whether to enable trace for the prompt.
        print_response: Whether to print the response to the console.
        newer_than: The timestamp to get trace results newer than.
            Defaults to None (all results).
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Example: Send a prompt within an existing session.
        Retrieve an existing session by providing the session ID and calling `get()`.
        Then, send a prompt to the agent.

            from synapseclient import Synapse
            from synapseclient.models.agent import AgentSession

            syn = Synapse()
            syn.login()

            my_session = AgentSession(id="foo").get()
            my_session.prompt(
                prompt="Hello",
                enable_trace=True,
                print_response=True,
            )
    """
    return AgentPrompt()

---

synapseclient.models.AgentPrompt dataclass

Bases: AsynchronousCommunicator

Represents a prompt, response, and metadata within an AgentSession.

ATTRIBUTE	DESCRIPTION
id	The unique ID of the agent prompt. TYPE: Optional[str]
session_id	The ID of the session that the prompt is associated with. TYPE: Optional[str]
prompt	The prompt to send to the agent. TYPE: Optional[str]
response	The response from the agent. TYPE: Optional[str]
enable_trace	Whether tracing is enabled for the prompt. TYPE: Optional[bool]
trace	The trace of the agent session. TYPE: Optional[str]

Source code in synapseclient/models/agent.py

@dataclass
class AgentPrompt(AsynchronousCommunicator):
    """Represents a prompt, response, and metadata within an AgentSession.

    Attributes:
        id: The unique ID of the agent prompt.
        session_id: The ID of the session that the prompt is associated with.
        prompt: The prompt to send to the agent.
        response: The response from the agent.
        enable_trace: Whether tracing is enabled for the prompt.
        trace: The trace of the agent session.
    """

    concrete_type: str = AGENT_CHAT_REQUEST

    id: Optional[str] = None
    """The unique ID of the agent prompt."""

    session_id: Optional[str] = None
    """The ID of the session that the prompt is associated with."""

    prompt: Optional[str] = None
    """The prompt sent to the agent."""

    response: Optional[str] = None
    """The response from the agent."""

    enable_trace: Optional[bool] = False
    """Whether tracing is enabled for the prompt."""

    trace: Optional[str] = None
    """The trace or "thought process" of the agent when responding to the prompt."""

    def to_synapse_request(self):
        """Converts the request to a request expected of the Synapse REST API."""
        return {
            "concreteType": self.concrete_type,
            "sessionId": self.session_id,
            "chatText": self.prompt,
            "enableTrace": self.enable_trace,
        }

    def fill_from_dict(self, synapse_response: Dict[str, str]) -> "AgentPrompt":
        """
        Converts a response from the REST API into this dataclass.

        Arguments:
            synapse_response: The response from the REST API.

        Returns:
            The AgentPrompt object.
        """
        self.id = synapse_response.get("jobId", None)
        self.session_id = synapse_response.get("sessionId", None)
        self.response = synapse_response.get("responseText", None)
        return self

    async def _post_exchange_async(
        self, *, synapse_client: Optional[Synapse] = None, **kwargs
    ) -> None:
        """Retrieves information about the trace of this prompt with the agent.

        Arguments:
            synapse_client: If not passed in and caching was not disabled by
                    `Synapse.allow_client_caching(False)` this will use the last created
                    instance from the Synapse class constructor.
        """
        if self.enable_trace:
            trace_response = await get_trace(
                prompt_id=self.id,
                newer_than=kwargs.get("newer_than", None),
                synapse_client=synapse_client,
            )
            self.trace = trace_response["page"][0]["message"]

Attributes

id class-attribute instance-attribute

id: Optional[str] = None

The unique ID of the agent prompt.

session_id class-attribute instance-attribute

session_id: Optional[str] = None

The ID of the session that the prompt is associated with.

prompt class-attribute instance-attribute

prompt: Optional[str] = None

The prompt sent to the agent.

response class-attribute instance-attribute

response: Optional[str] = None

The response from the agent.

enable_trace class-attribute instance-attribute

enable_trace: Optional[bool] = False

Whether tracing is enabled for the prompt.

trace class-attribute instance-attribute

trace: Optional[str] = None

The trace or "thought process" of the agent when responding to the prompt.

Functions

send_job_and_wait_async async

send_job_and_wait_async(post_exchange_args: Optional[Dict[str, Any]] = None, *, synapse_client: Optional[Synapse] = None) -> AsynchronousCommunicator

Send the job to the Asynchronous Job service and wait for it to complete. Intended to be called by a class inheriting from this mixin to start a job in the Synapse API and wait for it to complete. The inheriting class needs to represent an asynchronous job request and response and include all necessary attributes. This was initially implemented to be used in the AgentPrompt class which can be used as an example.

PARAMETER	DESCRIPTION
post_exchange_args	Additional arguments to pass to the request. TYPE: Optional[Dict[str, Any]] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
AsynchronousCommunicator	An instance of this class.

Using this function

This function was initially implemented to be used in the AgentPrompt class to send a prompt to an AI agent and wait for the response. It can also be used in any other class that needs to use an Asynchronous Job.

The inheriting class (AgentPrompt) will typically not be used directly, but rather through a higher level class (AgentSession), but this example shows how you would use this function.

from synapseclient import Synapse
from synapseclient.models.agent import AgentPrompt

syn = Synapse()
syn.login()

agent_prompt = AgentPrompt(
    id=None,
    session_id="123",
    prompt="Hello",
    response=None,
    enable_trace=True,
    trace=None,
)
# This will fill the id, response, and trace
# attributes with the response from the API
agent_prompt.send_job_and_wait_async()

Source code in synapseclient/models/mixins/asynchronous_job.py

async def send_job_and_wait_async(
    self,
    post_exchange_args: Optional[Dict[str, Any]] = None,
    *,
    synapse_client: Optional[Synapse] = None,
) -> "AsynchronousCommunicator":
    """Send the job to the Asynchronous Job service and wait for it to complete.
    Intended to be called by a class inheriting from this mixin to start a job
    in the Synapse API and wait for it to complete. The inheriting class needs to
    represent an asynchronous job request and response and include all necessary attributes.
    This was initially implemented to be used in the AgentPrompt class which can be used
    as an example.

    Arguments:
        post_exchange_args: Additional arguments to pass to the request.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        An instance of this class.

    Example: Using this function
        This function was initially implemented to be used in the AgentPrompt class
        to send a prompt to an AI agent and wait for the response. It can also be used
        in any other class that needs to use an Asynchronous Job.

        The inheriting class (AgentPrompt) will typically not be used directly, but rather
        through a higher level class (AgentSession), but this example shows how you would
        use this function.

            from synapseclient import Synapse
            from synapseclient.models.agent import AgentPrompt

            syn = Synapse()
            syn.login()

            agent_prompt = AgentPrompt(
                id=None,
                session_id="123",
                prompt="Hello",
                response=None,
                enable_trace=True,
                trace=None,
            )
            # This will fill the id, response, and trace
            # attributes with the response from the API
            agent_prompt.send_job_and_wait_async()
    """
    result = await send_job_and_wait_async(
        request=self.to_synapse_request(),
        request_type=self.concrete_type,
        synapse_client=synapse_client,
    )
    self.fill_from_dict(synapse_response=result)
    await self._post_exchange_async(
        **post_exchange_args, synapse_client=synapse_client
    )
    return self

to_synapse_request

to_synapse_request()

Converts the request to a request expected of the Synapse REST API.

Source code in synapseclient/models/agent.py

def to_synapse_request(self):
    """Converts the request to a request expected of the Synapse REST API."""
    return {
        "concreteType": self.concrete_type,
        "sessionId": self.session_id,
        "chatText": self.prompt,
        "enableTrace": self.enable_trace,
    }

fill_from_dict

fill_from_dict(synapse_response: Dict[str, str]) -> AgentPrompt

Converts a response from the REST API into this dataclass.

PARAMETER	DESCRIPTION
synapse_response	The response from the REST API. TYPE: Dict[str, str]

RETURNS	DESCRIPTION
AgentPrompt	The AgentPrompt object.

Source code in synapseclient/models/agent.py

def fill_from_dict(self, synapse_response: Dict[str, str]) -> "AgentPrompt":
    """
    Converts a response from the REST API into this dataclass.

    Arguments:
        synapse_response: The response from the REST API.

    Returns:
        The AgentPrompt object.
    """
    self.id = synapse_response.get("jobId", None)
    self.session_id = synapse_response.get("sessionId", None)
    self.response = synapse_response.get("responseText", None)
    return self

---