Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
A checklist to guide the planning of your BDK Book project, covering key requirements and considerations.
Ensure you have access to the following software requirements.
Books are collections of procedures that define specific actions within an automation. Each represents an automation task or step contributing to a larger process.
Designing a Book involves defining these procedures to ensure they align with the automation’s objectives. To guide this process, take into account the following considerations:
Identify the specific procedures needed for the Book. Each procedure should have a clear and specific automation purpose.
What actions does the user need to automate in Kognitos?
Can the actions be broken down into individual steps?
Consider these procedures for a Weather Book:
Get the current temperature in a specific city.
Check the current air quality index (AQI).
Determine delays or disruptions caused by severe weather for shipping and transportation.
Notify attendees of weather-related changes for events.
Clearly define each procedure's inputs, outputs, and processing steps.
What input or parameters are needed for each procedure?
What output or data should each procedure return upon completion?
How will data be transformed, manipulated, or processed in each procedure?
Consider an procedure that gets the temperature in a specific city:
Consider common error scenarios and define how they will be handled.
What are the potential error scenarios that could occur?
How will these errors be handled and/or logged?
Consider the following error scenarios for a Weather Book:
1. Invalid Input
A user enters a non-existent city name (e.g., "Atlantis").
The system receives a blank input for the city name.
2. API Unavailability
The weather API is temporarily down or unreachable.
A network timeout occurs during an API call.
If you plan to integrate with an external API in your Book, it’s important to assess the target service's requirements, limitations, and authentication mechanisms. The following considerations will ensure your chosen API aligns with your automation needs and integrates seamlessly into your BDK project.
Understand the authentication requirements for your target integration. BDK supports standard API authentication methods, including API keys, OAuth 2.0, client credentials, etc.
Does the API require authentication? If so, what kind?
Are there specific permissions or roles required to access certain API endpoints?
Can you obtain the necessary access and permissions for all required credentials?\
Identify which API endpoints will be used and define the data that will be exchanged.
Which API endpoints will your Book use?
What data will be sent or received?
Review the API’s rate limits and usage quotes. Ensure they can handle your expected request volume and check for additional costs or restrictions.
Are there API rate limits or usage quotes?
Does the API’s rate limit support the expected request volume for your workflows?
Are there additional costs or restrictions for exceeding usage limits?
Evaluate if the API documentation is clear and if provider support is accessible.
Is the API documentation comprehensive and easy to follow?
Does the API provider offer support?
Plan for future scalability and consider how the API will handle growing request needs.
Can the API handle increasing request volumes as workflows grow?
Does the API support additional functionality that might be required in future Books?
Prepare to deploy and manage a Docker container
Ensure the required infrastructure is in place to deploy a Docker container in the cloud, on-premises, or in a hybrid environment.
Plan strategies for:
Managing the container lifecycle, including scaling, networking, and security.
Implementing logging to track container activity
Monitoring container health and performance
Updating Docker images for new releases with clear versioning and testing practices.
Input
City name
Output
Temperature (°F)
Timestamp when the temperature was recorded
Processing Steps
Validate the city name.
Use an API request to get weather data for the specified city.
Parse the response to extract the temperature value and timestamp.
Format the temperature reading to the appropriate unit (Fahrenheit).
Learn about writing connection commands for Books built with BDK.
You must first learn your Book before connecting it.
A connection command is written in Kognitos to instantiate a connection to a third-party tool or service from a BDK Book. Whenever a Book implements a connection using the decorator, a connection command is required in your automation process.
Connection commands are also required when calling a from a BDK Book that requires a connection. Procedures requiring a connection will have connection_required=True in the decorator.
Use the following syntax to write a connection command in a Kognitos automation:
{book name} - The name of your Book, as defined in your decorator.
{noun phrase} – The noun_phrase defined in your decorator.
{preposition} – Any preposition that aligns with the plurality of the noun phrase. For example:
To pass arguments to your connection, add the with keyword at the end of the syntax. List each argument on a separately indented line, using the following format:
In this example, api keys is the noun phrase, and several connection arguments are provided.
NOT
0
Logical NOT operator, negates a condition.
Plural noun phrase: via some API keys
Singular noun phrase: via a client credentials method
connect {book name} via {preposition} {noun phrase}connect {book name} via {preposition} {noun phrase} with
the {argument 1} is "{value 1}"
the {argument 2} is "{value 2}"
the {argument 3} is "{value 3}"connect grimoire via the magic methodconnect photon via some api keys with
the api key is "1234-5678-ABCD-EFGH"
the client id is "client-98765"
the secret key is "s3cr3tK3y!"
the username is "photon_user"
the password is "P@ssw0rd!"Learn how to connect with third-party tools and services in your BDK project.
Connections enable you to connect to third-party tools and services in your custom Book. To implement a connection, define a Python function in your Book class and decorate it with the decorator. This decorated function serves as the connection handler and defines the syntax for writing a .
name
str
Optional
Specifies the name of the configuration. If not provided, the function name or value from *args is used.
default_value
Any
Optional
Specifies the default value for the configuration. If not provided, the default is None.
name
str
A name to associate with the connection. If not provided, it defaults to the function name.
noun_phrase
str
A unique label to identify the authentication method. If not provided, it will be inferred from the function name. Examples include, but are not limited to:
noun_phrase="api keys"
noun_phrase="client credentials"
noun_phrase="user password authentication"
This is an example implementation of connecting to the OpenWeather API using an API key:
Refer to connections for additional context and usage examples.
Note: This decorator always applies to the same method signature, but the method itself can have any name.
@oauthtoken
def function_name(self, access_token: str, expires_in: Optional[int] = None) -> None:You can define multiple connections within a Book, each with its own handler method. Use the noun_phrase keyword argument in the @connect decorator to assign a unique label to each connection and differentiate between authentication methods.
In your connection method docstring, include the following sections in addition to a brief summary:
Specify the Python function's parameters.
Define labels for the connection arguments (e.g., API key, credentials) that will be used in your connection command. Labels correspond to parameters in your function's definition.
When a label is provided in the docstring, the lowercase form is used in a connection command. When not provided, the label will be inferred from the Python variable name.
Example
In this example, API Key is the label for the api_key connection argument. The lowercase version of the label is used in a connection command (api key):
Additional Examples
api_key: ApI kEy
In this example, the connection command uses the lowercase form of ApI kEy:
api_key: aPi_KeY
In this example, the connection command uses the lowercase version of aPi_KeY:
No Label
If a label is not provided in the docstring, it is inferred from the Python variable name, api_key:
The NounPhase class can be used to represent a noun phrase.
A modifier is an optional part of a noun phrase that provides additional detail about the head noun. It appears before the head and refines the fact being referenced.
In this example:
Head: "dog"
Modifiers: "big", "white"
The resulting noun phrase is "big white dog".
Parameters can be defined as NounPhrase types to tell the system to use a fact's name instead of its value.
Consider the following procedure method definition, where the city parameter is defined as a NounPhrase:
This procedure can be called in an automation where London is a fact with a specific value:
In this example, when London is specified as the city parameter, the system uses the name of the fact (London) instead of its value ("windy").
DEFAULT_TIMEOUT = 30
@property
@config(default_value=DEFAULT_TIMEOUT)
def timeout(self) -> float:
"""
Timeout in seconds when making API calls.
"""
return self._timeout
@timeout.setter
def timeout(self, timeout: float):
"""
Sets the timeout value in seconds.
"""
if timeout <= 0:
raise ValueError("timeout must be positive")
self._timeout = timeout@connect(*args, **kwargs)@connect(noun_phrase="api keys")
def connect(self, api_key: str):
"""
Authenticate to Open Weather API using the specified API key.
You can obtain your own API key by visiting Open Weather's
website at https://openweathermap.org/appid.
Arguments:
api_key: The API key to be used for connecting
Labels:
api_key: API Key
"""
api_key = os.getenv("API_KEY", api_key)
test_url = f"{self._base_url}?appid={api_key}&q=London"
response = requests.get(test_url, timeout=self._timeout)
if response.status_code == 401:
response_data = response.json()
if "Invalid API key" in response_data.get("message", ""):
raise ValueError("Invalid API key")
self._api_key = api_key@oauth(
id="oauth",
provider=OAuthProvider.MICROSOFT,
flows=[OAuthFlow.AUTHORIZATION_CODE],
authorize_endpoint="https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/authorize",
token_endpoint="https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token",
scopes=[
"https://graph.microsoft.com/Files.ReadWrite",
"https://graph.microsoft.com/Sites.ReadWrite.All",
],
)
@book(name="MicrosoftBook")
class BaseMicrosoftBook:
"""
Base class for all Microsoft Books.
OAuth Arguments:
TENANT_ID: The tenant ID of the Azure AD directory.
OAuth Labels:
TENANT_ID: Tenant ID
"""
_acquired_token: Optional[AcquiredToken] = None
_headers: Optional[dict] = None
def __init__(self):
"""
Initializes an instance of the class.
"""
self._acquired_token = None
self._headers = None
@oauthtoken
def handle_token(self, access_token: str, expires_in: Optional[int] = None) -> None:
"""
Handles the OAuth token.
Args:
access_token (str): The OAuth access token.
expires_in (Optional[int]): The expiration time of the token, in seconds.
Labels:
access_token: OAuth Access Token
expires_in: Token Expiration Time
"""
self._acquired_token = AcquiredToken(access_token, expires_in or 3600)
@connect(noun_phrase="api keys")
def connect(self, api_key: str):
"""
Authenticate to Open Weather API using the specified API key. You can obtain your own API key by visiting
Open Weather's website at https://openweathermap.org/appid.
Arguments:
api_key: The API key to be used for connecting
Labels:
api_key: API Key
"""
api_key = os.getenv("API_KEY", api_key)
test_url = f"{self._base_url}?appid={api_key}&q=London"
response = requests.get(test_url, timeout=self._timeout)
if response.status_code == 401:
response_data = response.json()
if "Invalid API key" in response_data.get("message", ""):
raise ValueError("Invalid API key")
self._api_key = api_keyconnect openweather via some api keys method with
the api key is "aBcDefg1234"connect openweather via some api keys method with
the api key is "aBcDefg1234"connect openweather via some api keys method with
the api_key is "aBcDefg1234"connect openweather via some api keys method with
the api key is "aBcDefg1234"# Create a noun phrase with a head and modifiers
np = NounPhrase("dog", ["big", "white"])
print(np) # Outputs: big white dog@procedure("to get the (current temperature) at a city")
def current_temperature(
self, city: NounPhrase, unit: Optional[NounPhrase] = NounPhrase("metric")
) -> float:London is "windy"
get the current temperature at Londonid
str
A unique identifier for the Book.
icon
str
Path to an icon representing the Book. If not provided, a default icon will be used.
name
str
The name of the Book. If not provided, the name will be inferred from the class name.
noun_phrase
str
The book noun phrase.
If an icon is not specified, this will be used as the default.
AND
0
Logical AND operator, used to combine conditions.
OR
1
Logical OR operator, used to combine alternative conditions.
EQUALS
2
Tests if two values are equal.
NOT_EQUALS
3
Tests if two values are not equal.
IN
4
Tests if a value is within a specified list or range.
HAS
Learn how to implement a procedure with the BDK.
To implement a procedure in a BDK project, define a Python function in your Book class and decorate it with the @procedure decorator.
Ensure the name of your procedure adheres to the syntax guidelines specified in the name parameter of the decorator.
Your method should include the following sections:
A brief summary of the procedure
Input Concepts
Output Concepts
Examples are not required but are valuable for generating usage documentation.
Concepts and parameters must match to ensure they are properly mapped internally. Ensure your method definition adheres to the .
A singularized call is a way to call a procedure by phrasing it as if it returns a single item, even though it returns a list by definition. A procedure supports singularized calls in addition to standard calls if it meets all of the following conditions:
Returns a list.
The output of the procedure is the object itself.
The output noun phrase is plural.
The procedure accepts filters.
You don't need to implement additional logic for singular calls. BDK will automatically generate the singularized variation of any procedure that meets the above criteria.
Consider a procedure that retrieves users from Outlook. It can be called in two ways:
Standard Way: get some users from outlook whose whose mail is "example.com"
Singularized Way: get a user from outlook whose whose mail is "example.com"
@book(*args, **kwargs)
class BookClass:
"""
A sample book.
Author:
Kognitos
"""
# Book implementation here@book(name="Twilio", icon="data/twilio.svg")
class TwilioBook:
"""
The Twilio Book enables users to interact with the Twilio API.
"""
...from enum import Enum
class FilterBinaryOperator(Enum):
"""Represents the different binary operators used in filtering procedures."""
AND = 0
OR = 1
EQUALS = 2
NOT_EQUALS = 3
IN = 4
HAS = 5
LESS_THAN = 6
GREATER_THAN = 7
LESS_THAN_OR_EQUAL = 8
GREATER_THAN_OR_EQUAL = 95
Tests if a collection contains a specific element.
LESS_THAN
6
Tests if one value is less than another.
GREATER_THAN
7
Tests if one value is greater than another.
LESS_THAN_OR_EQUAL
8
Tests if one value is less than or equal to another.
GREATER_THAN_OR_EQUAL
9
Tests if one value is greater than or equal to another.
The BDK API is the core Python library in the Book Development Kit (BDK). It enables developers to build a Book. This reference covers the essential elements for working with the BDK API.
authorize_endpoint
str
Yes
The authorization endpoint URL for the OAuth provider.
token_endpoint
str
Yes
The token endpoint URL for exchanging the authorization code for a token.
scopes
List[str]
Optional
A list of scopes required by the OAuth provider for the authenticated requests.
id
str
Yes
The unique identifier for the OAuth configuration.
provider
OAuthProvider
Yes
The OAuth provider:
OAuthProvider.MICROSOFT
OAuthProvider.GOOGLE
flows
List[OAuthFlow]
Optional
A list of OAuth flows.
@oauth(
id=str,
provider=OAuthProvider,
flows=Optional[List[OAuthFlow]] = None,
authorize_endpoint=str,
token_endpoint=str,
scopes=Optional[List[str]] = None,
)
class ClassName:
"""
Class description
"""
# Class implementation@oauth(
id="oauth",
provider=OAuthProvider.MICROSOFT,
flows=[OAuthFlow.AUTHORIZATION_CODE],
authorize_endpoint="https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/authorize",
token_endpoint="https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token",
scopes=[
"https://graph.microsoft.com/Files.ReadWrite",
"https://graph.microsoft.com/Sites.ReadWrite.All",
],
)
@book(name="MicrosoftBook")
class BaseMicrosoftBook@procedure("to get some (users) from *office365*")
async def retrieve_users(self, filter_expression: Optional[FilterExpression] = None) -> List[OfficeUser]:
"""
Retrieves Office 365 users accessible via the Microsoft Graph API.
It requires the following permissions on the application:
User.Read.All, User.ReadWrite.All, Directory.Read.All, Directory.ReadWrite.All
Returns:
A list of Office 365 users.
Example 1:
Retrieve all users
>>> get users from office365
Example 2:
Retrieve a user whose email matches the specified email address
>>> get users from office365 whose mail is "[email protected]"
"""An overview of the Book Development Kit (BDK).
The Book Development Kit (BDK) is a software toolkit used to develop books for Kognitos. are collections of related automation procedures (tasks). They can be added to an agent to expand automation functionality and integrate with external tools.
Although Kognitos builds and maintains a of books to address common use cases, there may be situations that require a custom book. Using the BDK, you can develop, deploy, and connect your own book to help with:
With the BDK, you can create procedures that interact with third-party APIs, databases, and external services. If your organization uses tools or services not natively integrated with Kognitos, you can build a custom Book to integrate them directly into your automation workflows.
Extend Kognitos’ features to support more complex scenarios beyond standard functionality. Whether you need to support advanced data manipulation or custom data validation, the BDK provides the tools to create custom Books that meet your unique requirements.
The BDK is comprised of the following software components:
The Book Development Lifecycle outlines the stages of software development for a new Book and how the BDK components interact throughout the process.
The planning phase is where the Book's functionality is defined. In this stage, you identify the external systems the Book will integrate with and outline the workflows or automation goals. This stage clarifies objectives and requirements, setting the foundation for development.
Refer to our to help you plan your project.
The development phase is where the Book’s functionality is implemented using the BDK API and other tools:
The BDK Book Template provides starter code to streamline setup.
The BDK Linter enforces coding standards and identifies potential issues.
The BDK Poetry Plugin manages dependencies, versioning, and documentation.
The testing phase ensures that Books perform as intended. The BDK Book Template includes a testing framework for validating Book functionality.
In this stage, the Book is packaged and prepared for deployment. The code is packaged with the BDK Runtime using Docker and deployed externally.
Once deployed, the Book is ready to run in Kognitos, enabling you to extend your automations with custom workflows.
Learn how to set up your Book project with the BDK.
Ensure the following prerequisites are met and dependencies are installed before proceeding.
1. Python 3.11+
2.
3.
4.
5.
6.
Set up your project using the BDK Template, a tool designed to simplify the creation of Books.
Clone the from GitHub and initialize your project with Cookiecutter:
Enter the following project details or press Enter to keep the defaults:
Navigate into the new project directory. Replace project_slug with your own Project Slug from the previous step:
We recommend creating a virtual environment to isolate and manage your project dependencies.
1. to create the virtual environment inside the project’s root directory:
2. Create a virtual environment:
3. Activate the virtual environment
Run poetry shell or manually using the following commands:
Note: To later exit the virtual environment later, run
deactivate.
4. Install dependencies
Learn the requirements for deploying a BDK Book.
After successfully building your Book and creating a Docker image, the next step is deployment. You have the flexibility to deploy your Book to any environment that supports external access and routing, ensuring it can be reached from Kognitos. The choice of deployment method will depend on your organization's specific preferences and requirements. Common deployment options include:
Cloud platforms (ex: Amazon ECS, Azure)
On-premises infrastructure
Container orchestration platforms, such as Kubernetes
Port Configuration
Ensure that the port specified in the Dockerfile is properly exposed and matches your deployment environment's configuration.
Refer to our for additional deployment considerations.
Learn about implementing filter expressions in your BDK project.
Initial Version
Initial project version.
0.1.0
Project Name
The name of your project.
Sample Book
Project Slug
A URL-friendly version of the project name. If not provided, this will be derived from the project name.
sample_book
Project Description
A short description of the project.
A short description of the project.
Author Name
You or your organization's name.
pyenv install 3.11
Kognitos
is_a
str, List[str]
The name of the custom concept to be referred to in the Kognitos platform.*
unset
Any
Specifies a default value to represent an unset state for the concept.
*In the case of a List, each element would correspond to a noun phrase (a noun and its adjectives) in the hierarchy of possessives. For example:
is_a=["red dragon", "big house", "wooden door"]= red dragons's big house's wooden door
The following example demonstrates the usage of the @concept decorator to define the custom concept sms message:
To implement a filter expression, you need to provide the special filter_expression parameter in your procedure method definition. For example:
The FilterExpressionVisitor class is an abstract base class that defines methods for visiting different types of filter expressions. Each type of filter expression (binary, unary, value, noun phrase) is defined as a subclass of FilterExpression. You will need to define a class that implements these methods to handle filter expressions. For example:
Below is an example implementation of the FilterExpressionVisitor class in the Twilio Book:
Once you’ve defined your visitor class, you need to pass an instance of it to the filter expression. This is done by calling accept onfilter_expression. For example:
In this example, a filter expression is used in the read some SMS messages procedure:
cookiecutter git+ssh://[email protected]/kognitos/bdk-template.gitcd <project_slug>poetry config virtualenvs.in-project truepoetry env use 3.11pyenv: python3.11: command not found
The python3.11' command exists in these Python versions:
3.11.11pyenv local 3.11source $(poetry env info --path)/bin/activateInvoke-Expression (poetry env activate)poetry install@concept(*args, **kwargs)@concept(is_a="sms message")
@dataclass
class SMSMessage:
"""
An SMS (Short Message Service) message. Represents a text communication sent over a cellular network,
typically between mobile phones.
Attributes:
body: The text content of the message
num_segments: The number of segments that make up the complete message. SMS message bodies that exceed
the [character limit](https://www.twilio.com/docs/glossary/what-sms-character-limit) are segmented
and charged as multiple messages. Note: For messages sent via a Messaging Service, `num_segments` is
initially `0`, since a sender hasn't yet been assigned
sender_number: The sender's phone number (in [E.164](https://en.wikipedia.org/wiki/E.164) format),
[alphanumeric sender ID](https://www.twilio.com/docs/sms/quickstart),
[Wireless SIM](https://www.twilio.com/docs/iot/wireless/programmable-wireless-send-machine-machine-sms-commands),
[short code](https://www.twilio.com/en-us/messaging/channels/sms/short-codes), or
[channel address](https://www.twilio.com/docs/messaging/channels) (e.g., `whatsapp:+15554449999`). For
incoming messages, this is the number or channel address of the sender. For outgoing messages, this value
is a Twilio phone number, alphanumeric sender ID, short code, or channel address from which the
message is sent
recipient_number: The recipient's phone number (in [E.164](https://en.wikipedia.org/wiki/E.164) format) or
[channel address](https://www.twilio.com/docs/messaging/channels) (e.g. `whatsapp:+15552229999`)
date_updated: The [RFC 2822](https://datatracker.ietf.org/doc/html/rfc2822#section-3.3) timestamp (in GMT) of
when the Message resource was last updated
price: The amount billed for the message in the currency specified by `price_unit`. The `price` is populated
after the message has been sent/received, and may not be immediately availalble. View
the [Pricing page](https://www.twilio.com/en-us/pricing) for more details.
account_sid: The SID of the [Account](https://www.twilio.com/docs/iam/api/account) associated with
the Message resource
num_media: The number of media files associated with the Message resource.
status: The status of the message, for more information about possible statuses see
[Message Status](https://www.twilio.com/docs/messaging/api/message-resource#message-status-values)
messaging_service_sid: The SID of the [Messaging Service](https://www.twilio.com/docs/messaging/api/service-resource)
associated with the Message resource. A unique default value is assigned if a Messaging Service is not used.
sid: The unique, Twilio-provided string that identifies the Message resource.
date_sent: The [RFC 2822](https://datatracker.ietf.org/doc/html/rfc2822#section-3.3) timestamp (in GMT) of when
the Message was sent. For an outgoing message, this is when Twilio sent the message. For an
incoming message, this is when Twilio sent the HTTP request to your incoming message webhook URL.
date_created: The [RFC 2822](https://datatracker.ietf.org/doc/html/rfc2822#section-3.3) timestamp (in GMT) of
when the Message resource was created
price_unit: The currency in which `price` is measured, in
[ISO 4127](https://www.iso.org/iso/home/standards/currency_codes.htm) format (e.g. `usd`, `eur`, `jpy`).
error_message: The description of the `error_code` if the Message `status` is `failed` or `undelivered`. If no
error was encountered, the value is `null`.
error_code: The [error code](https://www.twilio.com/docs/api/errors) returned if the Message `status` is
`failed` or `undelivered`. If no error was encountered, the value is `null`.
"""
sid: Optional[str]
body: Optional[str]
num_segments: Optional[str]
sender_number: Optional[str]
recipient_number: Optional[str]
price: Optional[float]
account_sid: str
num_media: int
status: Optional[str]
messaging_service_sid: str
date_sent: Optional[datetime]
date_created: Optional[datetime]
date_updated: Optional[datetime]
price_unit: str
error_code: Optional[int]
error_message: Optional[str]
@classmethod
def from_message_instance(cls, message_instance: MessageInstance) -> Self:
"""
Constructs an SMSMessage object from a given MessageInstance object.
Arguments:
message_instance (MessageInstance): The MessageInstance object to create the SMSMessage from.
Returns:
The newly created SMSMessage object.
"""
return SMSMessage(
sid=message_instance.sid,
body=message_instance.body,
num_segments=message_instance.num_segments,
sender_number=message_instance.from_,
recipient_number=message_instance.to,
date_updated=message_instance.date_updated,
price=message_instance.price,
status=message_instance.status,
messaging_service_sid=message_instance.messaging_service_sid,
date_sent=message_instance.date_sent,
date_created=message_instance.date_created,
account_sid=message_instance.account_sid,
num_media=message_instance.num_media,
price_unit=message_instance.price_unit,
error_code=message_instance.error_code,
error_message=message_instance.error_message,
)get users from office365 whose email is "[email protected]"@procedure("to read some (*SMS* messages)")
def read_sms_messages(
self,
offset: Optional[int],
limit: Optional[int],
filter_expression: Optional[FilterExpression],
) -> List[SMSMessage]:
"""
Read some SMS messages using the Twilio API.
"""class MyFilterVisitor(FilterExpressionVisitor):
def visit_binary_expression(self, expression: FilterBinaryExpression) -> T:
# Implement logic for handling binary expressions
pass
def visit_unary_expression(self, expression: FilterUnaryExpression) -> T:
# Implement logic for handling unary expressions
pass
def visit_value(self, expression: ValueExpression) -> T:
# Implement logic for handling value expressions
pass
def visit_noun_phrases(self, expression: NounPhrasesExpression) -> T:
# Implement logic for handling noun phrases expressions
passSENDER_NUMBER = NounPhrase.from_word_list(["sender", "number"])
RECIPIENT_NUMBER = NounPhrase.from_word_list(["recipient", "number"])
DATE_SENT = NounPhrase.from_word_list(["date", "sent"])
class SMSMessageFilter(FilterExpressionVisitor):
"""
Extract filtering information from the expression
"""
current_noun_phrase: Optional[NounPhrase] = None
current_value: Optional[Any] = None
recipient_number: Union[str, object] = values.unset
sender_number: Union[str, object] = values.unset
date_sent: Union[datetime, object] = values.unset
date_sent_before: Union[datetime, object] = values.unset
date_sent_after: Union[datetime, object] = values.unset
def visit_binary_expression(self, expression: FilterBinaryExpression):
expression.left.accept(self)
expression.right.accept(self)
if expression.operator == FilterBinaryOperator.EQUALS:
if self.current_noun_phrase == SENDER_NUMBER:
self.sender_number = str(self.current_value)
elif self.current_noun_phrase == RECIPIENT_NUMBER:
self.recipient_number = str(self.current_value)
elif self.current_noun_phrase == DATE_SENT:
if not isinstance(self.current_value, datetime):
raise TypeMismatchError("date_sent", ConceptScalarType.DATETIME)
self.date_sent = self.current_value
elif expression.operator == FilterBinaryOperator.GREATER_THAN:
if self.current_noun_phrase == DATE_SENT:
if not isinstance(self.current_value, datetime):
raise TypeMismatchError("date_sent", ConceptScalarType.DATETIME)
self.date_sent_after = self.current_value
elif expression.operator == FilterBinaryOperator.LESS_THAN:
if self.current_noun_phrase == DATE_SENT:
if not isinstance(self.current_value, datetime):
raise TypeMismatchError("date_sent", ConceptScalarType.DATETIME)
self.date_sent_before = self.current_value
elif expression.operator == FilterBinaryOperator.AND:
pass
else:
raise ValueError(f"unsupported filtering operator: {expression.operator}")
def visit_unary_expression(self, expression: FilterUnaryExpression):
pass
def visit_value(self, expression: ValueExpression):
self.current_value = expression.value
def visit_noun_phrases(self, expression: NounPhrasesExpression):
if len(expression.noun_phrases) != 1:
raise ValueError(
f"unsupported filtering noun phrase: {expression.noun_phrases}"
)
if expression.noun_phrases[0] not in [
SENDER_NUMBER,
RECIPIENT_NUMBER,
DATE_SENT,
]:
raise ValueError(
f"unsupported filtering noun phrase: {expression.noun_phrases}"
)
self.current_noun_phrase = expression.noun_phrases[0]visitor = MyFilterVisitor()
filter_expression.accept(visitor)@procedure("to read some (*SMS* messages)")
def read_sms_messages(
self,
offset: Optional[int],
limit: Optional[int],
filter_expression: Optional[FilterExpression],
) -> List[SMSMessage]:
"""
Read some SMS messages using the Twilio API.
Returns:
A list of SMS messages that matches the specified filtering criteria
Example 1:
Retrieve SMS messages filtered by sender and recipient numbers
>>> read some sms messages whose sender number is "+18004445555" and whose recipient number is "+18004446666"
Example 2:
Retrieve SMS messages filtered by the date in which they were sent
>>> convert "2022-03-01T15:00:00Z" to a datetime
>>> use the above as the message date
>>> read some sms messages whose date sent is the message date
Example 3:
Retrieve SMS messages that were sent in the specified time period
>>> convert "2022-03-01T15:00:00Z" to a datetime
>>> use the above as the start date
>>> convert "2022-03-03T15:00:00Z" to a datetime
>>> use the above as the end date
>>> read some sms messages whose date sent is after the start date and whose date sent is before the end date
"""
client = Client(self._account_sid, self._auth_token)
filter_visitor = SMSMessageFilter()
if filter_expression is not None:
filter_expression.accept(filter_visitor)
messages = client.messages.list(
to=filter_visitor.recipient_number,
from_=filter_visitor.sender_number,
date_sent=filter_visitor.date_sent,
date_sent_before=filter_visitor.date_sent_before,
date_sent_after=filter_visitor.date_sent_after,
)
if offset is not None:
messages = messages[offset:]
if limit is not None:
messages = messages[:limit]
return [SMSMessage.from_message_instance(message) for message in messages]5
BDK Runtime
A runtime that connects a Book with the Kognitos platform. It is packaged as a Docker image that serves as the foundation for Book containers.
1
BDK API
The core Python library for Book development.
2
BDK Book Template
A Python Cookiecutter template that simplifies the process of developing a new Book.
3
BDK Linter
A linting tool built as a Pylint plugin. It analyzes code throughout development to identify potential issues and ensures Books are implemented, documented, and annotated correctly.
4
BDK Poetry Plugin
A Python dependency management and packaging tool that uses Poetry. It manages dependencies, versioning, formatting, and documentation generation for a Book.
1. Naming Conventions
Names must begin with to. This defines the action or intent of the procedure. For example:
@procedure("to capitalize a (string)"
@procedure("to get the (current temperature)")
@procedure("to send an *SMS* message")
2. Output Concepts
Output concepts are wrapped in parentheses (). For example:
3. Proper Nouns
Proper nouns are wrapped between asterisks **. For example:
In this example, office365 is considered a proper noun. The procedure is referred to as 'get some users from office365' rather than the office365.
name
str
Yes
A description that reflects the action or purpose of the procedure. See the syntax for details.
connection_required
bool
A boolean that indicates whether a connection to the service is required to execute the procedure. If not specified, it defaults to None.
noun_phrase
str
A string that represents the noun phrase for the procedure.
This method implements a procedure that capitalizes a string with one input concept and one output concept.
This method implements a procedure that creates an order in Truckmate. It has one input concept and two output concepts.
The following method implements a procedure that reads SMS messages using the Twilio API. In this example, SMS is a proper noun.
Output concepts represent the results that a procedure generates.
Input and output concepts can be defined as either standard or custom types.
The following Python data types can be used to define concepts:
Text
str
Number
float, int
Boolean
bool
Bytes
bytes
Date
datetime.datetime, datetime.date, datetime.time
File
typing.IO
Concepts can also be custom types. Custom concepts must be marked with the @concept decorator. Using a dataclass is recommended. For example:
Concepts and parameters are two different entities that are closely related.
Concepts are operated on by Kognitos procedures. Some concepts are included in the name of the @procedure decorator.
Parameters are operated on by Python functions. They are defined in a function's signature and are specific to the function's implementation.
An internal mapping is created between a Python function and a Kognitos procedure. To ensure a correct mapping, concepts and parameters must match.
Concepts and parameter names must match to ensure they are properly mapped internally.
Match concepts to parameters by following these guidelines:
Replace spaces with underscores.
Drop possessive constructions ('s).
Consider each connectednoun phrase as a separate parameter. Non-essential noun phrases (e.g., "field," "value") included to clarify context do not require a corresponding parameter.
Don't map articles ('an', 'a', 'the') to parameters; only the nouns they connect should be considered.
Define optional parameters in cases where the input concept is not explicitly defined in the procedure name.
In this example, the concept red car maps to the parameter red_car. The space is replaced with an underscore.
In the example below:
servicenow's ticket maps to ticket
field maps to field
outlook's standard user maps to standard_user
In this example, the concept ticket maps to the parameter ticket. The possessive construct ('s) is dropped and the word "field" is ignored.
In the example below, the concept of priority is not explicitly stated in the procedure name. It maps to the optional function parameter, priority.
In this example:
the city maps to the parameter city
the unit maps to the optional parameter the unit
The output concept, current temperature, is wrapped in parentheses in the procedure name.
Understand how to format and write docstrings to ensure your Book is well-documented.
BDK docstrings adhere to Google style formatting:
Docstrings are placed immediately after function, class, or method definitions.
A brief summary that describes the functionality begins the docstring.
The summary is followed by organized sections, separated by blank lines and labeled with headers.
In BDK projects, docstrings are applied to decorated methods and classes.
The following sections are supported in BDK docstrings, with alternative headers available for labeling flexibility.
Documents the arguments or parameters that a function or class accepts.
Supported Headers: Args, Arguments, Parameters, Params
Example
Documents the return value of a function or method.
Supported Headers: Returns
Example
Lists any exceptions or errors that a function might raise.
Supported Headers: Raises, Exceptions, Except
Example
List instance attributes in a class.
Supported Headers: Attributes
Example
Attributes the author(s) of the code.
Supported Headers: Author
Example
Defines labeled data associated with a function related to .
Supported Headers: Labels
Example
Describes OAuth labels for classes decorated with the .
Supported Headers: OAuth Labels
Example
Describes OAuth arguments for classes decorated with the .
Supported Headers: OAuth Arguments
Example
Describes the for a procedure.
Supported Headers: Input Concepts
Example
Describes the for a procedure.
Supported Headers: Output Concepts
Example
Outlines usage examples for a procedure.
Supported Headers: Example, Examples, Example [1-5]
Example
@procedure(name: str, **kwargs)@procedure("to capitalize a (string)")@procedure("to get some (users) from *office365*")@procedure("to capitalize a (string)", connection_required=False)
def capitalize_string(self, string: str) -> str:
"""
Capitalizes the input string.
Input Concepts:
the string: The string value you want to capitalize.
Output Concepts:
the string: The capitalized string.
Example 1:
Capitalize the string "hello"
>>> capitalize "hello"
"""
return string.capitalize()@procedure("to create an order in truckmate and get the order number and the bill number")
def create_order(self, order: OrderRequestConcept) -> Tuple[int, str]:
"""
Create a new order in Truckmate.
Input Concepts:
the order: The Truckmate order request (acc to openAPI spec)
Output Concepts:
the order number: Order ID of the created order
the bill number: Bill Number of the created order
Raises:
ValueError: If the order is not created successfully.
Example 1:
Create the order in Truckmate
>>> create a json
>>> use the above as the order
>>> set the order's id to 1234
>>> set the order's title to "Awesome title"
>>> set order's body to "This is the body"
>>> create the order in truckmate and get the order number and bill number
"""
logger.info("Creating Order in Truckmate")
post_body = OrdersPostRequest(orders=[order])
response = post_orders.sync_detailed(
client=self._client,
body=post_body,
)
if response.status_code == 201:
orders_response = response.parsed.to_dict() # type: ignore
if orders_response.get("orders"):
order = orders_response.get("orders")[0] # type: ignore
return order.get("orderId"), order.get("billNumber")
error_response = response.parsed.to_dict() # type: ignore
logger.error(
"Error while creating order in truckmate: %s",
error_response.get("errorText"),
)
raise ValueError(error_response)@procedure("to read some (*SMS* messages)")
def read_sms_messages(
self,
offset: Optional[int],
limit: Optional[int],
filter_expression: Optional[FilterExpression],
) -> List[SMSMessage]:
"""
Read some SMS messages using the Twilio API.
Returns:
A list of SMS messages that matches the specified filtering criteria
Example 1:
Retrieve SMS messages filtered by sender and recipient numbers
>>> read some sms messages whose sender number is "+18004445555" and whose recipient number is "+18004446666"
Example 2:
Retrieve SMS messages filtered by the date in which they were sent
>>> convert "2022-03-01T15:00:00Z" to a datetime
>>> use the above as the message date
>>> read some sms messages whose date sent is the message date
Example 3:
Retrieve SMS messages that were sent in the specified time period
>>> convert "2022-03-01T15:00:00Z" to a datetime
>>> use the above as the start date
>>> convert "2022-03-03T15:00:00Z" to a datetime
>>> use the above as the end date
>>> read some sms messages whose date sent is after the start date and whose date sent is before the end date
"""
client = Client(self._account_sid, self._auth_token)
filter_visitor = SMSMessageFilter()
if filter_expression is not None:
filter_expression.accept(filter_visitor)
messages = client.messages.list(
to=filter_visitor.recipient_number,
from_=filter_visitor.sender_number,
date_sent=filter_visitor.date_sent,
date_sent_before=filter_visitor.date_sent_before,
date_sent_after=filter_visitor.date_sent_after,
)
if offset is not None:
messages = messages[offset:]
if limit is not None:
messages = messages[:limit]
return [SMSMessage.from_message_instance(message) for message in messages]@concept(is_a="office user")
@dataclass
class OfficeUser:
"""
An Office User represents a user in the Microsoft Graph. It includes key user details such as display name,
email address, and job title.
Attributes:
id: The unique identifier for the user.
display_name: The name displayed in the address book for the user.
email_address: The user's email address (usually their user principal name).
job_title: The user's job title.
"""
id: str
display_name: Optional[str] = None
email_address: Optional[str] = None
job_title: Optional[str] = None
...@procedure("to start a red car")
def unique_function_name(self, red_car: type) -> output_type:@procedure("to send a servicenow's ticket's field to an outlook's standard user)
def func(self, ticket: Ticket, field: NounPhrase, standard_user: OutlookUser):@procedure("to update a ticket's field")
def update_field(self, ticket: Ticket, new_value: Any):@procedure("to assign the task to the user")
def assign_task(self, task: Task, user: User, priority: Optional[str] = None) -> None:
"""Assign a task to a user.
Input Concepts:
the task: The action of piece of work that needs to be completed.
the user: The user that the action is to be assigned to.
the priority: The level of importance or urgency of the task.
Example 1:
Assign the task to the user
>>> assign the task to the user
Example 2:
Assign the task to the user with a priority
>>> assign the task to the user with
the priority is "high"
"""@procedure("to get the (current temperature) at a city")
def current_temperature(self, city: NounPhrase, unit: Optional[NounPhrase] = NounPhrase("standard")) -> float:
"""Fetch the current temperature for a specified city.
Input Concepts:
the city: The name of the city.
the unit: Unit of measurement.
Output Concepts:
the current temperature: The current temperature in the specified units of measurement, or None if an error occurs.
Example 1:
Retrieve the current temperature at London
>>> get the current temperature at London
Example 2:
Retrieve the current temperature at London in Celsius
>>> get the current temperature at Buenos Aires with
... the unit is metric
"""UUID
uuid.UUID
Table
pyarrow.Table, arro3.core.Table, nanoarrow.ArrayStream
Lists
List[str], List[int]
Dictionary
Dict[str, Any]
Note:
Anycan be any of the other supported types.
@procedure("to add two numbers")
def add_numbers(a: int, b: int = 0) -> int:
"""
Adds two numbers together.
Args:
a: The first number.
b: The second number. Defaults to 0.
"""
return a + b@procedure("to send an *SMS* message")
def send_sms_message(self, sender_number: str, recipient_number: str, message_body: str) -> Optional[str]:
"""
Sends an SMS message using the Twilio API.
Returns:
The SID of the sent message if successful, otherwise None.
"""@timeout.setter
def timeout(self, timeout: float):
"""
Sets the timeout value in milliseconds.
Args:
timeout (int): The timeout value to set. Must be a positive integer.
Raises:
ValueError: If the timeout value is less than or equal to 0.
"""
if timeout <= 0:
raise ValueError("timeout must be positive")
self._timeout = timeout@concept(is_a="office user")
@dataclass
class OfficeUser:
"""
An Office User represents a user in the Microsoft Graph. It includes key user details such as display name,
email address, and job title.
Attributes:
id: The unique identifier for the user.
display_name: The name displayed in the address book for the user.
email_address: The user's email address (usually their user principal name).
job_title: The user's job title.
"""
id: str
display_name: Optional[str] = None
email_address: Optional[str] = None
job_title: Optional[str] = None@book(name="Sample Book")
class SampleBook:
"""
A book shown as an example.
Author:
Kognitos
"""@connect(noun_phrase="api keys")
def connect(self, api_key: str):
"""
Authenticate to Open Weather API using the specified API key.
Labels:
api_key: API Key
"""@oauth(
id="oauth",
provider=OAuthProvider.MICROSOFT,
flows=[OAuthFlow.AUTHORIZATION_CODE, OAuthFlow.CLIENT_CREDENTIALS],
authorize_endpoint="https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/authorize",
token_endpoint="https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token",
scopes=[
"https://graph.microsoft.com/Files.ReadWrite",
"https://graph.microsoft.com/Sites.Manage.All",
"https://graph.microsoft.com/Sites.ReadWrite.All",
"https://graph.microsoft.com/User.ReadWrite.All",
"https://graph.microsoft.com/Directory.ReadWrite.All",
"https://graph.microsoft.com/Group.ReadWrite.All",
"https://graph.microsoft.com/GroupMember.ReadWrite.All",
"https://graph.microsoft.com/Mail.ReadWrite",
"https://graph.microsoft.com/Mail.ReadWrite.Shared",
"https://graph.microsoft.com/Mail.Send",
"https://graph.microsoft.com/Calendars.ReadWrite",
],
)
@book(name="MicrosoftBook")
class BaseMicrosoftBook:
"""
Base class for all Microsoft Books.
OAuth Arguments:
TENANT_ID: The tenant ID of the Azure AD directory.
OAuth Labels:
TENANT_ID: Tenant ID
"""@oauth(
id="oauth",
provider=OAuthProvider.MICROSOFT,
flows=[OAuthFlow.AUTHORIZATION_CODE, OAuthFlow.CLIENT_CREDENTIALS],
authorize_endpoint="https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/authorize",
token_endpoint="https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token",
scopes=[
"https://graph.microsoft.com/Files.ReadWrite",
"https://graph.microsoft.com/Sites.Manage.All",
"https://graph.microsoft.com/Sites.ReadWrite.All",
"https://graph.microsoft.com/User.ReadWrite.All",
"https://graph.microsoft.com/Directory.ReadWrite.All",
"https://graph.microsoft.com/Group.ReadWrite.All",
"https://graph.microsoft.com/GroupMember.ReadWrite.All",
"https://graph.microsoft.com/Mail.ReadWrite",
"https://graph.microsoft.com/Mail.ReadWrite.Shared",
"https://graph.microsoft.com/Mail.Send",
"https://graph.microsoft.com/Calendars.ReadWrite",
],
)
@book(name="MicrosoftBook")
class BaseMicrosoftBook:
"""
Base class for all Microsoft Books.
OAuth Arguments:
TENANT_ID: The tenant ID of the Azure AD directory.
OAuth Labels:
TENANT_ID: Tenant ID
"""@procedure("to get the (current temperature) at a city")
def current_temperature(self, city: NounPhrase, unit: Optional[NounPhrase] = NounPhrase("metric")) -> float:
"""
Fetch the current temperature for a specified city.
Input Concepts:
the city: The name of the city. Please refer to ISO 3166 for the state codes or country codes.
the unit: Unit of measurement. standard, metric and imperial units are available. If you do
not specify the units, metric units will be applied by default.
"""@procedure("to get the (current temperature) at a city")
def current_temperature(self, city: NounPhrase, unit: Optional[NounPhrase] = NounPhrase("metric")) -> float:
"""
Fetch the current temperature for a specified city.
Output Concepts:
the current temperature: The current temperature in the specified units of measurement, or None if an error occurs.
"""@procedure("to get the (current temperature) at a city")
def current_temperature(self, city: NounPhrase, unit: Optional[NounPhrase] = NounPhrase("metric")) -> float:
"""
Fetch the current temperature for a specified city.
Example 1:
Retrieve the current temperature at London
>>> get the current temperature at London
Example 2:
Retrieve the current temperature at London in Celsius
>>> get the current temperature at Buenos Aires with
... the unit is metric
"""
Learn how to implement questions in a BDK Book.
In your Book, you may need to ask the user for information. For example, if the Book needs to know when a project should begin, it can ask the user, “What is the project start date?” Or if it needs a username, it can prompt the user with, “What is the username?”
When this happens, Kognitos surfaces a question to pause the automation and ask the user for information. A question is an exception that signals that the automation cannot move forward without user input. The BDK provides an easy way to add and handle questions in your Book using the ask() method.
The ask() function in the BDK is used to ask a question. Here’s how it works:
To use ask() in your Book, follow these steps:
This example shows how to use ask() with a set of mixed-type choices: a string, a datetime, and a float. The return type allows for any of those types, and also accounts for the case where a Question is returned.
Sometimes your Book may need to collect more than one piece of information. You can call ask() multiple times in the same procedure. Here’s an example:
The first call to ask() checks whether the answer for color is present.
If not, it returns a Question.
If answered, it stores the value and restarts the procedure.
Now that the answer for color is present, the procedure checks for the
A guide to learning a custom BDK Book in Kognitos.
Before an agent can use a Book’s procedures in a Playground or Process, it must first learn the Book. This allows the agent to recognize the Book and its procedures.
Follow the steps below to learn your custom Book after deployment.
Retrieve the HTTPS endpoint from your deployment.
Place this at the beginning of your automation, substituting your own endpoint in the syntax:
If you implemented any custom , make sure you also to your Book after learning.
learn "https://<your-endpoint>"Each time ask() is called, Kognitos checks internally whether it has already stored an answer for that specific question before.
If the answer to the question exists, ask() returns the answer value itself.
If the answer to the question does not exist, ask() returns a Question.
When a Question object is returned from a procedure, Kognitos will:
Forward the question to the user
Pause the automation and wait for a response
Store the answer internally
Restart the procedure from the beginning automatically
The second time the procedure runs, ask() will be able to return the answer instead of a Question.
Call the ask() function inside a method decorated with @procedure.
Provide the concept_name (the unique name of the concept being asked for) and the concept_type (the return type of the answer).
(Optional) Specifying the Question Text
You can specify a text to present to the user when the question is prompted in Kognitos. If the text is not specified, questions are presented as Please provide the <concept_name> by default. For example:
(Optional) Specifying Choices
You can also guide users to select an answer from a predefined list by passing a choices argument. When choices are set, Kognitos displays the question with a dropdown menu for the user to select from. For example:
The ask() function can return either the answer or a Question object. To handle both cases, explicitly check the return type:
If ask() returns a value, that means the answer already exists internally.
If ask() returns a Question, that means the answer does not exist internally. What you do with the Question is up to you — the Book can return it, hold onto it for later, or ignore it entirely. If you want Kognitos to surface the question to the user, return the Question object from the procedure. For example:
Always include return types in your procedure definitions. If your procedure may trigger a Question, the return type must reflect that so the system understands which concept(s) you may be asking for. Make sure to include both the concept name (as a string Literal in nounphrase form) and the concept type in the type hint. For example:
If not, it returns a Question.
If answered, it stores the value and restarts the procedure.
Once both values are available, the final result is returned.
concept_name
Yes
The unique name of the concept being asked for.
concept_type
Yes
The return type of the answer.
text
No
(Optional) A message to show the user when the question is prompted in Kognitos.
choices
No
(Optional) A list of predefined options the user can choose from.
result = ask("project start date", str)if isinstance(start_date := ask("project start date", str), Question):
return start_date
# Use the answer as a datetime object
date_obj = datetime.strptime(date_str, "%Y-%m-%d")
project_due_date = date_obj + timedelta(days=30)@procedure("to get a project due (date)")
def get_due_date(self) -> str | Question[Literal["project start date"], str]:ask(concept_name: NounPhrasesStringLiteral, concept_type: Type[T], text: Optional[str] = None, choices: Optional[List[T]] = None) -> Union[T, Question[NounPhrasesStringLiteral, T]]from kognitos.bdk.api.questions import ask, Question@procedure("to get a (value)")
def get_value(self) -> str | datetime | float | Question[Literal["value"], str | datetime | float]:
choices = ["Hello!", datetime(2023, 1, 1, tzinfo=timezone.utc), 123.456]
if isinstance(answer := ask("value", Union[str, datetime, float], choices=choices), Question):
return answer
return answerfrom typing import Literal
from kognitos import procedure
from kognitos.bdk.api.questions import ask, Question
@procedure("to get a (text)")
def get_text(self) -> str | Question[Literal["color"], str] | Question[Literal["elephant's name"], str]:
"""
Get a text
"""
# First: Ask the user to pick a color
if isinstance(answer := ask("color", str, text="Pick a color", choices=["red", "blue", "green"]), Question):
return answer
color = answer
# Second: Ask for the elephant's name
if isinstance(answer := ask("elephant's name", str, text="What is the elephant's name?"), Question):
return answer
name = answer
# You can now use both answers
return f"The elephant named {name} is painted {color}."result = ask("project start date", str, text="What is the start date of the project?")result = ask("project start date", str, choices=["2025-07-22", "2025-07-25", "2025-07-28"])if isinstance(result := ask("project start date", str), Question)
return resultLearn how to develop, package, and run your Book locally with the BDK.
Have You Completed the Setup?
Ensure you’ve followed all the steps to set up your BDK project before proceeding. This includes configuring your environment and installing dependencies.
The project is organized with the following structure:
The project root directory is named using the Project Slug (default: sample_book).
📃 book.py: Contains the core class where your Book is defined and implemented.
📁 data: Holds images like 🖼️ icon.svg, the default SVG used as a Book's icon.
📃 test_book.py: Contains unit tests to validate the Book's functionality.
📃 Dockerfile: Builds the Docker image.
📃 pyproject.toml: Manages dependencies and settings.
📃 poetry_scripts.py: Custom automation script.
📃 README.md: Project overview documentation, including setup and usage instructions.
📃 USAGE.md: Reference documentation for your Book. This file is not included by default. For details on how to generate it, see .
Implement your Book in book.py. This class is auto-generated by the BDK Template and serves as the starting point for your Book’s implementation. Use decorators to configure settings, define procedures, and establish API connections. Refer to the BDK API Reference and for details.
is used for Python dependency management in BDK projects. To add an external Python dependency to your project, refer to the command, which adds packages to your pyproject.toml and installs them.
This section outlines essential commands for Book development and testing.
Code formatting automatically adjusts your code's structure, ensuring adherence to coding standards, readability, and maintainability. To apply the formatting, run:
is used as the source linter for BDK projects. The linter analyzes and enforce coding standards, checks for errors in Python code, and improves code quality. To run the linter:
This command will generate a comprehensive USAGE.md file by extracting and formatting the from your code. To generate the documentation, run:
is used to test and validate your book. To run the test suite:
Build a Docker image to package your Book:
Architecture Compatibility
This image pulls the BDK Runtime base image from Docker Hub, which is currently available only for the linux/amd64 architecture. If you're using a machine with a different architecture, you can use the --platform flag to emulate linux/amd64 when building the image:
To test your Book in Kognitos before deployment, you can run your Docker image locally with . Ngrok is a tool that creates a secure tunnel to your local machine and provides a public URL to make your local Docker image accessible from the platform.
1. Install ngrok
Follow the installation steps for based on your system. You will need to sign up for a free account.
2. Obtain your ngrok authtoken
Navigate to Your Authtoken in the ngrok portal and copy your token.
3. Add your authtoken
4. Configure the ngrok api key as an environment variable:
5. Build & run in ngrok mode
This command will invoke a custom script that builds the Docker image and runs it in ngrok mode:
If you are running into issues with `poetry run host`, you can and run your book manually with the following command:
Make sure to replace auth_token, project_slug, and version with your own values.
6. Copy the URL
You'll see the ngrok address in the following format in the logs. Copy the URL:
7. Add the URL to your automation
Copy the URL and paste it into your Kognitos Playground using this syntax:
If you implemented any custom , make sure you also to your Book after learning.
📃 poetry.lock: Locks dependency versions.
📂 sample_book
├── 📁 src
│ └── 📁 sample_book
│ ├── 📃 __init__.py
│ ├── 📃 __version__.py
│ └── 📃 book.py
│ └── 📁 data
│ └── 🖼️ icon.svg
├── 📁 tests
│ └── 📃 test_book.py
├── 📃 Dockerfile
├── 📃 pyproject.toml
├── 📃 poetry_scripts.py
├── 📃 poetry.lock
├── 📃 README.md
└── 📃 USAGE.mdpoetry run formatpoetry run lintpoetry run docpoetry run testsdocker build -t <project_slug>:<version> .docker build --platform linux/amd64 -t <project_slug>:<version> .ngrok config add-authtoken <token>export NGROK_AUTHTOKEN=<token>poetry run hostdocker run -e BDK_SERVER_MODE=ngrok -e NGROK_AUTHTOKEN=<auth_token> <project_slug>:<version>listening on https://<SOME_UUID>.ngrok-free.applearn "https://f1d7-100-34-252-18.ngrok-free.app"