ToolSet Server

graphorchestrator.toolsetserver.runtime

StateModel

Bases: BaseModel

Represents the state model for messages.

Attributes:

Name	Type	Description
messages	List[Any]	A list to store messages of any type.

Source code in graphorchestrator\toolsetserver\runtime.py

class StateModel(BaseModel):
    """
    Represents the state model for messages.

    Attributes:
        messages (List[Any]): A list to store messages of any type.
    """

    messages: List[Any] = Field(default_factory=list)

ToolSetServer

Base class for creating a tool set server.

This class handles the creation of FastAPI endpoints for registered tool methods.

Source code in graphorchestrator\toolsetserver\runtime.py

class ToolSetServer(metaclass=_ToolSetMeta):
    """
    Base class for creating a tool set server.

    This class handles the creation of FastAPI endpoints for registered tool methods.
    """

    host: str = "127.0.0.1"
    port: int = 8000
    name: str = "ToolSet"
    require_auth: bool = False

    @classmethod
    # Class method to handle authentication.
    def authenticate(cls, token: str) -> bool:
        """
        Authenticates a request based on the provided token.

        Args:
            token (str): The authentication token.

        Returns:
            bool: True if the token is valid, False otherwise.
        """
        return False

    @classmethod
    def serve(cls, **uvicorn_kwargs: Any):
        """
        Starts the FastAPI server synchronously.

        Args:
            **uvicorn_kwargs: Keyword arguments to pass to uvicorn.run.

        Raises:
            RuntimeError: If there's an error during server start-up.
        """
        uvicorn.run(
            cls._fastapi,
            host=uvicorn_kwargs.pop("host", cls.host),
            port=uvicorn_kwargs.pop("port", cls.port),
            log_level=uvicorn_kwargs.pop("log_level", "info"),
            **uvicorn_kwargs,
        )

    @classmethod
    async def serve_async(cls, **uvicorn_kwargs: Any):
        """
        Starts the FastAPI server asynchronously.

        Args:
            **uvicorn_kwargs: Keyword arguments to pass to uvicorn.run.
        Raises:
            RuntimeError: If there's an error during server start-up.
        """
        config = uvicorn.Config(
            cls._fastapi,
            host=uvicorn_kwargs.pop("host", cls.host),
            port=uvicorn_kwargs.pop("port", cls.port),
            log_level=uvicorn_kwargs.pop("log_level", "info"),
            **uvicorn_kwargs,
        )
        server = uvicorn.Server(config)
        await server.serve()

authenticate(token) classmethod

Authenticates a request based on the provided token.

Parameters:

Name	Type	Description	Default
token	str	The authentication token.	required

Returns:

Name	Type	Description
bool	bool	True if the token is valid, False otherwise.

Source code in graphorchestrator\toolsetserver\runtime.py

@classmethod
# Class method to handle authentication.
def authenticate(cls, token: str) -> bool:
    """
    Authenticates a request based on the provided token.

    Args:
        token (str): The authentication token.

    Returns:
        bool: True if the token is valid, False otherwise.
    """
    return False

serve(**uvicorn_kwargs) classmethod

Starts the FastAPI server synchronously.

Parameters:

Name	Type	Description	Default
**uvicorn_kwargs	Any	Keyword arguments to pass to uvicorn.run.	{}

Raises:

Type	Description
RuntimeError	If there's an error during server start-up.

Source code in graphorchestrator\toolsetserver\runtime.py

@classmethod
def serve(cls, **uvicorn_kwargs: Any):
    """
    Starts the FastAPI server synchronously.

    Args:
        **uvicorn_kwargs: Keyword arguments to pass to uvicorn.run.

    Raises:
        RuntimeError: If there's an error during server start-up.
    """
    uvicorn.run(
        cls._fastapi,
        host=uvicorn_kwargs.pop("host", cls.host),
        port=uvicorn_kwargs.pop("port", cls.port),
        log_level=uvicorn_kwargs.pop("log_level", "info"),
        **uvicorn_kwargs,
    )

serve_async(**uvicorn_kwargs) async classmethod

Starts the FastAPI server asynchronously.

Parameters:

Name	Type	Description	Default
**uvicorn_kwargs	Any	Keyword arguments to pass to uvicorn.run.	{}

Raises: RuntimeError: If there's an error during server start-up.

Source code in graphorchestrator\toolsetserver\runtime.py

@classmethod
async def serve_async(cls, **uvicorn_kwargs: Any):
    """
    Starts the FastAPI server asynchronously.

    Args:
        **uvicorn_kwargs: Keyword arguments to pass to uvicorn.run.
    Raises:
        RuntimeError: If there's an error during server start-up.
    """
    config = uvicorn.Config(
        cls._fastapi,
        host=uvicorn_kwargs.pop("host", cls.host),
        port=uvicorn_kwargs.pop("port", cls.port),
        log_level=uvicorn_kwargs.pop("log_level", "info"),
        **uvicorn_kwargs,
    )
    server = uvicorn.Server(config)
    await server.serve()