Client

AsyncLlamaDeployClient

Source code in llama_deploy/llama_deploy/client/async_client.py

class AsyncLlamaDeployClient:
    def __init__(
        self, control_plane_config: ControlPlaneConfig, timeout: float = DEFAULT_TIMEOUT
    ):
        self.control_plane_config = control_plane_config
        # TODO: add scheme to config (http, https, ..)
        self.control_plane_url = control_plane_config.url
        self.timeout = timeout

    async def create_session(
        self, poll_interval: float = DEFAULT_POLL_INTERVAL
    ) -> AsyncSessionClient:
        """Create a new session and return a AsyncSessionClient for it.

        Returns:
            AsyncSessionClient: A client for the newly created session.
        """
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.post(f"{self.control_plane_url}/sessions/create")
            session_id = response.json()
        return AsyncSessionClient(
            self.control_plane_config,
            session_id,
            timeout=self.timeout,
            poll_interval=poll_interval,
        )

    async def list_sessions(self) -> List[SessionDefinition]:
        """List all sessions registered with the control plane.

        Returns:
            List[SessionDefinition]: A list of session definitions.
        """
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.get(f"{self.control_plane_url}/sessions")
            return [SessionDefinition(**session) for session in response.json()]

    async def get_session(
        self, session_id: str, poll_interval: float = DEFAULT_POLL_INTERVAL
    ) -> AsyncSessionClient:
        """Get an existing session by ID.

        Args:
            session_id (str): The ID of the session to get.

        Returns:
            AsyncSessionClient: A client for the specified session.

        Raises:
            ValueError: If the session does not exist.
        """
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.get(
                f"{self.control_plane_url}/sessions/{session_id}"
            )
            if response.status_code == 404:
                raise ValueError(f"Session with id {session_id} not found")
            response.raise_for_status()

        return AsyncSessionClient(
            self.control_plane_config,
            session_id,
            timeout=self.timeout,
            poll_interval=poll_interval,
        )

    async def get_or_create_session(
        self, session_id: str, poll_interval: float = DEFAULT_POLL_INTERVAL
    ) -> AsyncSessionClient:
        """Get an existing session by ID, or create a new one if it doesn't exist.

        Args:
            session_id (str): The ID of the session to get.

        Returns:
            AsyncSessionClient: A client for the specified session.
        """
        try:
            return await self.get_session(session_id, poll_interval=poll_interval)
        except ValueError as e:
            if "not found" in str(e):
                return await self.create_session(poll_interval=poll_interval)
            raise e

    async def get_service(self, service_name: str) -> ServiceDefinition:
        """Get the definition of a service by name.

        Args:
            service_name (str): The name of the service to get.

        Returns:
            ServiceDefinition: The definition of the service.
        """
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.get(
                f"{self.control_plane_url}/services/{service_name}"
            )
            return ServiceDefinition(**response.json())

    async def delete_session(self, session_id: str) -> None:
        """Delete a session by ID.

        Args:
            session_id (str): The ID of the session to delete.
        """
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            await client.post(f"{self.control_plane_url}/sessions/{session_id}/delete")

    async def list_services(self) -> List[ServiceDefinition]:
        """List all services registered with the control plane.

        Returns:
            List[ServiceDefinition]: A list of service definitions.
        """
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.get(f"{self.control_plane_url}/services")
            return [
                ServiceDefinition(**service) for _, service in response.json().items()
            ]

    async def register_service(self, service_def: ServiceDefinition) -> None:
        """Register a service with the control plane.

        Args:
            service_def (ServiceDefinition): The service definition to register.
        """
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            await client.post(
                f"{self.control_plane_url}/services/register",
                json=service_def.model_dump(),
            )

    async def deregister_service(self, service_name: str) -> None:
        """Deregister a service from the control plane.

        Args:
            service_name (str): The name of the service to deregister.
        """
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            await client.post(
                f"{self.control_plane_url}/services/deregister",
                json={"service_name": service_name},
            )

create_session async

create_session(poll_interval: float = DEFAULT_POLL_INTERVAL) -> AsyncSessionClient

Create a new session and return a AsyncSessionClient for it.

Returns:

Name	Type	Description
AsyncSessionClient	AsyncSessionClient	A client for the newly created session.

Source code in llama_deploy/llama_deploy/client/async_client.py

async def create_session(
    self, poll_interval: float = DEFAULT_POLL_INTERVAL
) -> AsyncSessionClient:
    """Create a new session and return a AsyncSessionClient for it.

    Returns:
        AsyncSessionClient: A client for the newly created session.
    """
    async with httpx.AsyncClient(timeout=self.timeout) as client:
        response = await client.post(f"{self.control_plane_url}/sessions/create")
        session_id = response.json()
    return AsyncSessionClient(
        self.control_plane_config,
        session_id,
        timeout=self.timeout,
        poll_interval=poll_interval,
    )

list_sessions async

list_sessions() -> List[SessionDefinition]

List all sessions registered with the control plane.

Returns:

Type	Description
List[SessionDefinition]	List[SessionDefinition]: A list of session definitions.

Source code in llama_deploy/llama_deploy/client/async_client.py

async def list_sessions(self) -> List[SessionDefinition]:
    """List all sessions registered with the control plane.

    Returns:
        List[SessionDefinition]: A list of session definitions.
    """
    async with httpx.AsyncClient(timeout=self.timeout) as client:
        response = await client.get(f"{self.control_plane_url}/sessions")
        return [SessionDefinition(**session) for session in response.json()]

get_session async

get_session(session_id: str, poll_interval: float = DEFAULT_POLL_INTERVAL) -> AsyncSessionClient

Get an existing session by ID.

Parameters:

Name	Type	Description	Default
session_id	str	The ID of the session to get.	required

Returns:

Name	Type	Description
AsyncSessionClient	AsyncSessionClient	A client for the specified session.

Raises:

Type	Description
ValueError	If the session does not exist.

Source code in llama_deploy/llama_deploy/client/async_client.py

async def get_session(
    self, session_id: str, poll_interval: float = DEFAULT_POLL_INTERVAL
) -> AsyncSessionClient:
    """Get an existing session by ID.

    Args:
        session_id (str): The ID of the session to get.

    Returns:
        AsyncSessionClient: A client for the specified session.

    Raises:
        ValueError: If the session does not exist.
    """
    async with httpx.AsyncClient(timeout=self.timeout) as client:
        response = await client.get(
            f"{self.control_plane_url}/sessions/{session_id}"
        )
        if response.status_code == 404:
            raise ValueError(f"Session with id {session_id} not found")
        response.raise_for_status()

    return AsyncSessionClient(
        self.control_plane_config,
        session_id,
        timeout=self.timeout,
        poll_interval=poll_interval,
    )

get_or_create_session async

get_or_create_session(session_id: str, poll_interval: float = DEFAULT_POLL_INTERVAL) -> AsyncSessionClient

Get an existing session by ID, or create a new one if it doesn't exist.

Parameters:

Name	Type	Description	Default
session_id	str	The ID of the session to get.	required

Returns:

Name	Type	Description
AsyncSessionClient	AsyncSessionClient	A client for the specified session.

Source code in llama_deploy/llama_deploy/client/async_client.py

async def get_or_create_session(
    self, session_id: str, poll_interval: float = DEFAULT_POLL_INTERVAL
) -> AsyncSessionClient:
    """Get an existing session by ID, or create a new one if it doesn't exist.

    Args:
        session_id (str): The ID of the session to get.

    Returns:
        AsyncSessionClient: A client for the specified session.
    """
    try:
        return await self.get_session(session_id, poll_interval=poll_interval)
    except ValueError as e:
        if "not found" in str(e):
            return await self.create_session(poll_interval=poll_interval)
        raise e

get_service async

get_service(service_name: str) -> ServiceDefinition

Get the definition of a service by name.

Parameters:

Name	Type	Description	Default
service_name	str	The name of the service to get.	required

Returns:

Name	Type	Description
ServiceDefinition	ServiceDefinition	The definition of the service.

Source code in llama_deploy/llama_deploy/client/async_client.py

async def get_service(self, service_name: str) -> ServiceDefinition:
    """Get the definition of a service by name.

    Args:
        service_name (str): The name of the service to get.

    Returns:
        ServiceDefinition: The definition of the service.
    """
    async with httpx.AsyncClient(timeout=self.timeout) as client:
        response = await client.get(
            f"{self.control_plane_url}/services/{service_name}"
        )
        return ServiceDefinition(**response.json())

delete_session async

delete_session(session_id: str) -> None

Delete a session by ID.

Parameters:

Name	Type	Description	Default
session_id	str	The ID of the session to delete.	required

Source code in llama_deploy/llama_deploy/client/async_client.py

async def delete_session(self, session_id: str) -> None:
    """Delete a session by ID.

    Args:
        session_id (str): The ID of the session to delete.
    """
    async with httpx.AsyncClient(timeout=self.timeout) as client:
        await client.post(f"{self.control_plane_url}/sessions/{session_id}/delete")

list_services async

list_services() -> List[ServiceDefinition]

List all services registered with the control plane.

Returns:

Type	Description
List[ServiceDefinition]	List[ServiceDefinition]: A list of service definitions.

Source code in llama_deploy/llama_deploy/client/async_client.py

async def list_services(self) -> List[ServiceDefinition]:
    """List all services registered with the control plane.

    Returns:
        List[ServiceDefinition]: A list of service definitions.
    """
    async with httpx.AsyncClient(timeout=self.timeout) as client:
        response = await client.get(f"{self.control_plane_url}/services")
        return [
            ServiceDefinition(**service) for _, service in response.json().items()
        ]

register_service async

register_service(service_def: ServiceDefinition) -> None

Register a service with the control plane.

Parameters:

Name	Type	Description	Default
service_def	ServiceDefinition	The service definition to register.	required

Source code in llama_deploy/llama_deploy/client/async_client.py

async def register_service(self, service_def: ServiceDefinition) -> None:
    """Register a service with the control plane.

    Args:
        service_def (ServiceDefinition): The service definition to register.
    """
    async with httpx.AsyncClient(timeout=self.timeout) as client:
        await client.post(
            f"{self.control_plane_url}/services/register",
            json=service_def.model_dump(),
        )

deregister_service async

deregister_service(service_name: str) -> None

Deregister a service from the control plane.

Parameters:

Name	Type	Description	Default
service_name	str	The name of the service to deregister.	required

Source code in llama_deploy/llama_deploy/client/async_client.py

async def deregister_service(self, service_name: str) -> None:
    """Deregister a service from the control plane.

    Args:
        service_name (str): The name of the service to deregister.
    """
    async with httpx.AsyncClient(timeout=self.timeout) as client:
        await client.post(
            f"{self.control_plane_url}/services/deregister",
            json={"service_name": service_name},
        )

LlamaDeployClient

Source code in llama_deploy/llama_deploy/client/sync_client.py

class LlamaDeployClient:
    def __init__(
        self, control_plane_config: ControlPlaneConfig, timeout: float = DEFAULT_TIMEOUT
    ):
        self.control_plane_config = control_plane_config
        # TODO: add scheme to config (http, https, ..)
        self.control_plane_url = control_plane_config.url
        self.timeout = timeout

    def create_session(
        self, poll_interval: float = DEFAULT_POLL_INTERVAL
    ) -> SessionClient:
        """Create a new session and return a SessionClient for it.

        Returns:
            SessionClient: A client for the newly created session.
        """
        with httpx.Client(timeout=self.timeout) as client:
            response = client.post(f"{self.control_plane_url}/sessions/create")
            session_id = response.json()
        return SessionClient(
            self.control_plane_config,
            session_id,
            timeout=self.timeout,
            poll_interval=poll_interval,
        )

    def list_sessions(self) -> List[SessionDefinition]:
        """List all sessions registered with the control plane.

        Returns:
            List[SessionDefinition]: A list of session definitions.
        """
        with httpx.Client(timeout=self.timeout) as client:
            response = client.get(f"{self.control_plane_url}/sessions")
            return [SessionDefinition(**session) for session in response.json()]

    def get_session(
        self, session_id: str, poll_interval: float = DEFAULT_POLL_INTERVAL
    ) -> SessionClient:
        """Get an existing session by ID.

        Args:
            session_id (str): The ID of the session to get.

        Returns:
            SessionClient: A client for the specified session.

        Raises:
            ValueError: If the session does not exist.
        """
        with httpx.Client(timeout=self.timeout) as client:
            response = client.get(f"{self.control_plane_url}/sessions/{session_id}")
            if response.status_code == 404:
                raise ValueError(f"Session with id {session_id} not found")
            response.raise_for_status()

        return SessionClient(
            self.control_plane_config,
            session_id,
            timeout=self.timeout,
            poll_interval=poll_interval,
        )

    def get_or_create_session(
        self, session_id: str, poll_interval: float = DEFAULT_POLL_INTERVAL
    ) -> SessionClient:
        """Get an existing session by ID, or create a new one if it doesn't exist.

        Args:
            session_id (str): The ID of the session to get.

        Returns:
            SessionClient: A client for the specified session.
        """
        try:
            return self.get_session(session_id, poll_interval=poll_interval)
        except ValueError as e:
            if "not found" in str(e):
                return self.create_session(poll_interval=poll_interval)
            raise e

    def get_service(self, service_name: str) -> ServiceDefinition:
        """Get the definition of a service by name.

        Args:
            service_name (str): The name of the service to get.

        Returns:
            ServiceDefinition: The definition of the service.
        """
        with httpx.Client(timeout=self.timeout) as client:
            response = client.get(f"{self.control_plane_url}/services/{service_name}")
            return ServiceDefinition(**response.json())

    def delete_session(self, session_id: str) -> None:
        """Delete a session by ID.

        Args:
            session_id (str): The ID of the session to delete.
        """
        with httpx.Client(timeout=self.timeout) as client:
            client.post(f"{self.control_plane_url}/sessions/{session_id}/delete")

    def list_services(self) -> List[ServiceDefinition]:
        """List all services registered with the control plane.

        Returns:
            List[ServiceDefinition]: A list of service definitions.
        """
        with httpx.Client(timeout=self.timeout) as client:
            response = client.get(f"{self.control_plane_url}/services")
            return [
                ServiceDefinition(**service) for _, service in response.json().items()
            ]

    def register_service(self, service_def: ServiceDefinition) -> None:
        """Register a service with the control plane.

        Args:
            service_def (ServiceDefinition): The service definition to register.
        """
        with httpx.Client(timeout=self.timeout) as client:
            client.post(
                f"{self.control_plane_url}/services/register",
                json=service_def.model_dump(),
            )

    def deregister_service(self, service_name: str) -> None:
        """Deregister a service from the control plane.

        Args:
            service_name (str): The name of the service to deregister.
        """
        with httpx.Client(timeout=self.timeout) as client:
            client.post(
                f"{self.control_plane_url}/services/deregister",
                json={"service_name": service_name},
            )

create_session

create_session(poll_interval: float = DEFAULT_POLL_INTERVAL) -> SessionClient

Create a new session and return a SessionClient for it.

Returns:

Name	Type	Description
SessionClient	SessionClient	A client for the newly created session.

Source code in llama_deploy/llama_deploy/client/sync_client.py

def create_session(
    self, poll_interval: float = DEFAULT_POLL_INTERVAL
) -> SessionClient:
    """Create a new session and return a SessionClient for it.

    Returns:
        SessionClient: A client for the newly created session.
    """
    with httpx.Client(timeout=self.timeout) as client:
        response = client.post(f"{self.control_plane_url}/sessions/create")
        session_id = response.json()
    return SessionClient(
        self.control_plane_config,
        session_id,
        timeout=self.timeout,
        poll_interval=poll_interval,
    )

list_sessions

list_sessions() -> List[SessionDefinition]

List all sessions registered with the control plane.

Returns:

Type	Description
List[SessionDefinition]	List[SessionDefinition]: A list of session definitions.

Source code in llama_deploy/llama_deploy/client/sync_client.py

def list_sessions(self) -> List[SessionDefinition]:
    """List all sessions registered with the control plane.

    Returns:
        List[SessionDefinition]: A list of session definitions.
    """
    with httpx.Client(timeout=self.timeout) as client:
        response = client.get(f"{self.control_plane_url}/sessions")
        return [SessionDefinition(**session) for session in response.json()]

get_session

get_session(session_id: str, poll_interval: float = DEFAULT_POLL_INTERVAL) -> SessionClient

Get an existing session by ID.

Parameters:

Name	Type	Description	Default
session_id	str	The ID of the session to get.	required

Returns:

Name	Type	Description
SessionClient	SessionClient	A client for the specified session.

Raises:

Type	Description
ValueError	If the session does not exist.

Source code in llama_deploy/llama_deploy/client/sync_client.py

def get_session(
    self, session_id: str, poll_interval: float = DEFAULT_POLL_INTERVAL
) -> SessionClient:
    """Get an existing session by ID.

    Args:
        session_id (str): The ID of the session to get.

    Returns:
        SessionClient: A client for the specified session.

    Raises:
        ValueError: If the session does not exist.
    """
    with httpx.Client(timeout=self.timeout) as client:
        response = client.get(f"{self.control_plane_url}/sessions/{session_id}")
        if response.status_code == 404:
            raise ValueError(f"Session with id {session_id} not found")
        response.raise_for_status()

    return SessionClient(
        self.control_plane_config,
        session_id,
        timeout=self.timeout,
        poll_interval=poll_interval,
    )

get_or_create_session

get_or_create_session(session_id: str, poll_interval: float = DEFAULT_POLL_INTERVAL) -> SessionClient

Get an existing session by ID, or create a new one if it doesn't exist.

Parameters:

Name	Type	Description	Default
session_id	str	The ID of the session to get.	required

Returns:

Name	Type	Description
SessionClient	SessionClient	A client for the specified session.

Source code in llama_deploy/llama_deploy/client/sync_client.py

def get_or_create_session(
    self, session_id: str, poll_interval: float = DEFAULT_POLL_INTERVAL
) -> SessionClient:
    """Get an existing session by ID, or create a new one if it doesn't exist.

    Args:
        session_id (str): The ID of the session to get.

    Returns:
        SessionClient: A client for the specified session.
    """
    try:
        return self.get_session(session_id, poll_interval=poll_interval)
    except ValueError as e:
        if "not found" in str(e):
            return self.create_session(poll_interval=poll_interval)
        raise e

get_service

get_service(service_name: str) -> ServiceDefinition

Get the definition of a service by name.

Parameters:

Name	Type	Description	Default
service_name	str	The name of the service to get.	required

Returns:

Name	Type	Description
ServiceDefinition	ServiceDefinition	The definition of the service.

Source code in llama_deploy/llama_deploy/client/sync_client.py

def get_service(self, service_name: str) -> ServiceDefinition:
    """Get the definition of a service by name.

    Args:
        service_name (str): The name of the service to get.

    Returns:
        ServiceDefinition: The definition of the service.
    """
    with httpx.Client(timeout=self.timeout) as client:
        response = client.get(f"{self.control_plane_url}/services/{service_name}")
        return ServiceDefinition(**response.json())

delete_session

delete_session(session_id: str) -> None

Delete a session by ID.

Parameters:

Name	Type	Description	Default
session_id	str	The ID of the session to delete.	required

Source code in llama_deploy/llama_deploy/client/sync_client.py

def delete_session(self, session_id: str) -> None:
    """Delete a session by ID.

    Args:
        session_id (str): The ID of the session to delete.
    """
    with httpx.Client(timeout=self.timeout) as client:
        client.post(f"{self.control_plane_url}/sessions/{session_id}/delete")

list_services

list_services() -> List[ServiceDefinition]

List all services registered with the control plane.

Returns:

Type	Description
List[ServiceDefinition]	List[ServiceDefinition]: A list of service definitions.

Source code in llama_deploy/llama_deploy/client/sync_client.py

def list_services(self) -> List[ServiceDefinition]:
    """List all services registered with the control plane.

    Returns:
        List[ServiceDefinition]: A list of service definitions.
    """
    with httpx.Client(timeout=self.timeout) as client:
        response = client.get(f"{self.control_plane_url}/services")
        return [
            ServiceDefinition(**service) for _, service in response.json().items()
        ]

register_service

register_service(service_def: ServiceDefinition) -> None

Register a service with the control plane.

Parameters:

Name	Type	Description	Default
service_def	ServiceDefinition	The service definition to register.	required

Source code in llama_deploy/llama_deploy/client/sync_client.py

def register_service(self, service_def: ServiceDefinition) -> None:
    """Register a service with the control plane.

    Args:
        service_def (ServiceDefinition): The service definition to register.
    """
    with httpx.Client(timeout=self.timeout) as client:
        client.post(
            f"{self.control_plane_url}/services/register",
            json=service_def.model_dump(),
        )

deregister_service

deregister_service(service_name: str) -> None

Deregister a service from the control plane.

Parameters:

Name	Type	Description	Default
service_name	str	The name of the service to deregister.	required

Source code in llama_deploy/llama_deploy/client/sync_client.py

def deregister_service(self, service_name: str) -> None:
    """Deregister a service from the control plane.

    Args:
        service_name (str): The name of the service to deregister.
    """
    with httpx.Client(timeout=self.timeout) as client:
        client.post(
            f"{self.control_plane_url}/services/deregister",
            json={"service_name": service_name},
        )

options: members: - LlamaDeployClient - AsyncLlamaDeployClient