Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions packages/gooddata-sdk/src/gooddata_sdk/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -274,6 +274,16 @@
from gooddata_sdk.compute.compute_to_sdk_converter import ComputeToSdkConverter
from gooddata_sdk.compute.model.attribute import Attribute
from gooddata_sdk.compute.model.base import ExecModelEntity, ObjId
from gooddata_sdk.compute.model.chat_context import (
CatalogActiveObjectIdentification,
CatalogDashboardContext,
CatalogObjectReference,
CatalogObjectReferenceGroup,
CatalogRichTextWidgetDescriptor,
CatalogUIContext,
CatalogUserContext,
CatalogWidgetDescriptor,
)
from gooddata_sdk.compute.model.execution import (
BareExecutionResponse,
Execution,
Expand Down
174 changes: 174 additions & 0 deletions packages/gooddata-sdk/src/gooddata_sdk/compute/model/chat_context.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,174 @@
# (C) 2026 GoodData Corporation
from __future__ import annotations

from typing import Any

import attrs
from gooddata_api_client.model.active_object_identification import ActiveObjectIdentification
from gooddata_api_client.model.dashboard_context import DashboardContext
from gooddata_api_client.model.object_reference import ObjectReference
from gooddata_api_client.model.object_reference_group import ObjectReferenceGroup
from gooddata_api_client.model.rich_text_widget_descriptor import RichTextWidgetDescriptor
from gooddata_api_client.model.ui_context import UIContext
from gooddata_api_client.model.user_context import UserContext
from gooddata_api_client.model.widget_descriptor import WidgetDescriptor

from gooddata_sdk.catalog.base import Base


@attrs.define(kw_only=True)
class CatalogObjectReference(Base):
"""Reference to a GoodData object by its ID and type."""

id: str
type: str # Allowed: WIDGET, METRIC, ATTRIBUTE, DASHBOARD

@staticmethod
def client_class() -> type[ObjectReference]:
return ObjectReference


@attrs.define(kw_only=True)
class CatalogObjectReferenceGroup(Base):
"""Group of explicitly referenced objects, optionally scoped by a context."""

objects: list[CatalogObjectReference] = attrs.field(factory=list)
context: CatalogObjectReference | None = None

@staticmethod
def client_class() -> type[ObjectReferenceGroup]:
return ObjectReferenceGroup

def as_api_model(self) -> ObjectReferenceGroup:
api_objects = [o.to_api() for o in self.objects]
kwargs: dict[str, Any] = {}
if self.context is not None:
kwargs["context"] = self.context.to_api()
return ObjectReferenceGroup(_check_type=False, objects=api_objects, **kwargs)


@attrs.define(kw_only=True)
class CatalogActiveObjectIdentification(Base):
"""Identifies the object currently active (open/selected) by the user."""

id: str
type: str
workspace_id: str

@staticmethod
def client_class() -> type[ActiveObjectIdentification]:
return ActiveObjectIdentification


@attrs.define(kw_only=True)
class CatalogWidgetDescriptor(Base):
"""Describes a widget currently visible on the dashboard."""

title: str
widget_id: str
widget_type: str
filters: list[dict[str, Any]] = attrs.field(factory=list)

@staticmethod
def client_class() -> type[WidgetDescriptor]:
return WidgetDescriptor

def as_api_model(self) -> WidgetDescriptor:
kwargs: dict[str, Any] = {}
if self.filters:
kwargs["filters"] = self.filters
return WidgetDescriptor(
_check_type=False,
title=self.title,
widget_id=self.widget_id,
widget_type=self.widget_type,
**kwargs,
)


@attrs.define(kw_only=True)
class CatalogRichTextWidgetDescriptor(Base):
"""Describes a rich text widget currently visible on the dashboard."""

title: str
widget_id: str
filters: list[dict[str, Any]] = attrs.field(factory=list)

@staticmethod
def client_class() -> type[RichTextWidgetDescriptor]:
return RichTextWidgetDescriptor

def as_api_model(self) -> RichTextWidgetDescriptor:
kwargs: dict[str, Any] = {}
if self.filters:
kwargs["filters"] = self.filters
return RichTextWidgetDescriptor(
_check_type=False,
title=self.title,
widget_id=self.widget_id,
**kwargs,
)


@attrs.define(kw_only=True)
class CatalogDashboardContext(Base):
"""Context describing the currently open dashboard and its visible widgets."""

id: str
widgets: list[CatalogWidgetDescriptor] = attrs.field(factory=list)

@staticmethod
def client_class() -> type[DashboardContext]:
return DashboardContext

def as_api_model(self) -> DashboardContext:
return DashboardContext(
_check_type=False,
id=self.id,
widgets=[w.as_api_model() for w in self.widgets],
)


@attrs.define(kw_only=True)
class CatalogUIContext(Base):
"""UI context describing the user's current view."""

dashboard: CatalogDashboardContext | None = None

@staticmethod
def client_class() -> type[UIContext]:
return UIContext

def as_api_model(self) -> UIContext:
kwargs: dict[str, Any] = {}
if self.dashboard is not None:
kwargs["dashboard"] = self.dashboard.as_api_model()
return UIContext(_check_type=False, **kwargs)


@attrs.define(kw_only=True)
class CatalogUserContext(Base):
"""User context for AI chat, providing information about the user's current state.

Pass an instance of this class to :py:meth:`ComputeService.ai_chat` or
:py:meth:`ComputeService.ai_chat_stream` to give the AI assistant richer
context about what the user is currently looking at.
"""

active_object: CatalogActiveObjectIdentification | None = None
referenced_objects: list[CatalogObjectReferenceGroup] = attrs.field(factory=list)
view: CatalogUIContext | None = None

@staticmethod
def client_class() -> type[UserContext]:
return UserContext

def as_api_model(self) -> UserContext:
kwargs: dict[str, Any] = {}
if self.active_object is not None:
kwargs["active_object"] = self.active_object.to_api()
if self.referenced_objects:
kwargs["referenced_objects"] = [r.as_api_model() for r in self.referenced_objects]
if self.view is not None:
kwargs["view"] = self.view.as_api_model()
return UserContext(_check_type=False, **kwargs)
31 changes: 27 additions & 4 deletions packages/gooddata-sdk/src/gooddata_sdk/compute/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
from gooddata_api_client.model.search_result import SearchResult

from gooddata_sdk.client import GoodDataApiClient
from gooddata_sdk.compute.model.chat_context import CatalogUserContext
from gooddata_sdk.compute.model.execution import (
Execution,
ExecutionDefinition,
Expand Down Expand Up @@ -135,17 +136,28 @@ def build_exec_def_from_chat_result(
is_cancellable=is_cancellable,
)

def ai_chat(self, workspace_id: str, question: str) -> ChatResult:
def ai_chat(
self,
workspace_id: str,
question: str,
*,
user_context: CatalogUserContext | None = None,
) -> ChatResult:
"""
Chat with AI in GoodData workspace.

Args:
workspace_id (str): workspace identifier
question (str): question for the AI
user_context (CatalogUserContext, optional): context about the user's current state,
used to provide richer answers (e.g. active dashboard, referenced objects).
Returns:
ChatResult: Chat response
"""
chat_request = ChatRequest(question=question)
chat_params: dict[str, Any] = {}
if user_context is not None:
chat_params["user_context"] = user_context.as_api_model()
chat_request = ChatRequest(question=question, **chat_params)
response = self._actions_api.ai_chat(workspace_id, chat_request, _check_return_type=False)
return response

Expand All @@ -160,17 +172,28 @@ def _parse_sse_events(self, raw: str) -> Iterator[Any]:
except json.JSONDecodeError:
continue

def ai_chat_stream(self, workspace_id: str, question: str) -> Iterator[Any]:
def ai_chat_stream(
self,
workspace_id: str,
question: str,
*,
user_context: CatalogUserContext | None = None,
) -> Iterator[Any]:
"""
Chat Stream with AI in GoodData workspace.

Args:
workspace_id (str): workspace identifier
question (str): question for the AI
user_context (CatalogUserContext, optional): context about the user's current state,
used to provide richer answers (e.g. active dashboard, referenced objects).
Returns:
Iterator[Any]: Yields parsed JSON objects from each SSE event's data field
"""
chat_request = ChatRequest(question=question)
chat_params: dict[str, Any] = {}
if user_context is not None:
chat_params["user_context"] = user_context.as_api_model()
chat_request = ChatRequest(question=question, **chat_params)
response = self._actions_api.ai_chat_stream(
workspace_id, chat_request, _check_return_type=False, _preload_content=False
)
Expand Down
Loading
Loading