Build a channel from scratch - Part 3: Connecting to the community

The new endpoints in your integration service need to connect to your Zendesk community to accomplish two tasks:

• Get the community posts created after a specific start time
• Add an agent comment in reply to a specific community post

The endpoints declared two functions to do this work:

• community.get_new_posts()
• community.create_post_comment()

In this article, you'll define these functions in a separate module called community.

Topics covered:

• Creating the community module
• Getting new posts
• Replying to new posts

This is the third part of a project to build a channel from scratch:

• Part 1: Getting started
• Part 2: Creating the integration service
• Part 3: Connecting to the community - YOU ARE HERE
• Part 4: Building the admin interface
• Part 5: Deploying the channel

Disclaimer: Zendesk provides this article for demonstration and instructional purposes only. The channel developed in the article should not be used in a production environment. Zendesk does not provide support for developing a channel.

Creating the community module

1. Create a file called .env and save it in the same folder as your service.py file.

   The .env file contains environment variables for your project. When you upload your web app to Glitch, secrets stored in the .env aren't publicly visible.

2. Paste the following into the .env file:

   ZENDESK_URL="{zendesk_url}"
   ZENDESK_EMAIL="{email}"
   ZENDESK_PASSWORD="{password}"
   ZENDESK_AGENT_ID={zendesk_user_id}

   Replace {zendesk_url} with your Zendesk account URL, such as https://example.zendesk.com. Replace {email} and {password} with your the email address and password for your Zendesk account.

   Replace {zendesk_user_id} with your Zendesk user id or the id of an agent in your account for this demo. This id is the agent you want to display as the author of the comment replying to the post. You'd normally make this value dynamic, but that's beyond the scope of this tutorial.

3. Create a file called community.py and save it in the same folder as your service.py file.

4. In community.py file, paste the following:

   import json
   import urllib.parse
   import arrow
   import requests
   import os
   from dotenv import load_dotenv
   
   load_dotenv()
   
   def get_hc_settings():
       return {
           'base_url': os.environ.get('ZENDESK_URL'),
           'credentials': (os.environ.get('ZENDESK_EMAIL'),
                           os.environ.get('ZENDESK_PASSWORD')),
           'agent_id': os.environ.get('ZENDESK_AGENT_ID')
       }

   The community is part of a Zendesk help center and is accessed using the Help Center API. The module's get_hc_settings function lets your integration service make authenticated calls to the API.

5. Switch to your service.py file and add the following import statement at the top of the file:

   import community
   ...

You've now given the endpoints in the service.py file access to the community module. The next step is to define the module functions for getting new posts and adding comments to posts.

Getting new posts

The Pull New Posts endpoint in service.py hands off the task of getting new community posts to a function called get_new_posts() in the community module:

new_posts = community.get_new_posts(topic_id, start_time)

The function takes two arguments:

• topic_id - the community topic where customers add posts to ask for help
• start_time - a datetime to start looking for new posts

To add the get_new_posts() function to the community module

Paste the following function into the community.py file and save the file.

...
def get_new_posts(topic_id, start_time):
    if start_time:
        start_time = arrow.get(start_time)
    else:
        start_time = arrow.utcnow().shift(hours=-1)

    hc = get_hc_settings()
    url = f'{hc["base_url"]}/api/v2/community/topics/{topic_id}/posts.json'
    headers = {'Content-Type': 'application/json'}

    # get new posts
    new_posts = []
    while url:
        response = requests.get(url, auth=hc['credentials'], headers=headers)
        if response.status_code != 200:
            error =  {'status_code': response.status_code, 'text': response.text}
            return {'error': error}
        data = response.json()
        url = data['next_page']
        for post in data['posts']:
            created_at_time = arrow.get(post['created_at'])
            if created_at_time > start_time:
                new_posts.append(post)
            else:                   # no more new posts
                url = None          # stop paginating
                break               # stop 'for' loop

    # reformat data for response
    external_resources = []
    for post in new_posts:
        external_id = str(post['author_id'])
        author = {'external_id': external_id, 'name': 'community user'}
        resource = {
            'external_id': str(post['id']),
            'message': post['title'],
            'html_message': post['details'],
            'created_at': post['created_at'],
            'author': author
        }
        external_resources.append(resource)

    # get next start time
    if external_resources:
        created_at_times = []
        for resource in external_resources:
            created_at_times.append(arrow.get(resource['created_at']))
        start_time = max(created_at_times)   # get most recent datetime in list

    # add start_time to state
    state = {'start_time': start_time.isoformat()}

    # convert state dict to JSON object, then URL encode
    state = urllib.parse.quote(json.dumps(state))

    return {
        'external_resources': external_resources,
        'state': state
    }

How it works

The function starts by checking if a start time is set:

def get_new_posts(topic_id, start_time):
    if start_time:
        start_time = arrow.get(state['start_time'])
    else:
        start_time = arrow.utcnow().shift(hours=-1)
        # one hour ago for first request

The start time, if set, is an ISO 8601 formatted string. The Arrow library's arrow.get() method converts it into a datetime object for easy comparison with other dates.

If the start_time is an empty string, it means the channel has just been deployed and the value hasn't been set yet. In that case, the function sets the start time to one hour in the past:

start_time = arrow.utcnow().shift(hours=-1)

Next, the function retrieves your help center settings and prepares to make a request to the List Posts endpoint in the Help Center API:

hc = get_hc_settings()
url = f'{hc["base_url"]}/api/v2/community/topics/{topic_id}/posts.json'
headers = {'Content-Type': 'application/json'}

Next, the function makes the request:

new_posts = []
while url:
    response = requests.get(url, auth=hc['credentials'], headers=headers)
    if response.status_code != 200:
        error = {'status_code': response.status_code, 'text': response.text}
        return {'error': error}
    data = response.json()
    url = data['next_page']
    for post in data['posts']:
        created_at_time = arrow.get(post['created_at'])
        if created_at_time > start_time:
            new_posts.append(post)
        else:                   # no more new posts
            url = None          # stop paginating
            break               # exit for loop

There's a lot to unpack here. Let's break it down.

If the results consist of more than one page ("while there's a URL for a next page"), the function loops and makes a request for the next page.

while url:

The function requests the first page of results:

response = requests.get(url, auth=hc['credentials'], headers=headers)
if response.status_code != 200:
    error = {'status_code': response.status_code, 'text': response.text}
    return {'error': error}

If the request doesn't return any errors, the function gets the data from the response and sets the next page url:

data = response.json()
url = data['next_page']

For each post in the results, the function checks to see if the post was created after the start time:

for post in data['posts']:
    created_at_time = arrow.get(post['created_at'])
    if created_at_time > start_time:
        new_posts.append(post)

If the post was created after the start time, the function appends it to the new_posts list.

By default, the posts in the results are sorted by the created_at time. Logically then, if you find a post older than the start time, you won't find any more new posts in the results and you can stop looking:

else:
    url = None          # stop paginating
    break               # exit for loop

Setting the next page url to None breaks out of the while url loop. The break keyword breaks out of the for loop.

You come out of it with a list of new posts in new_posts.

The next step is to package the post data in the format that Zendesk expects, which is documented in external_resources array in the dev docs.

external_resources = []
for post in new_posts:
    resource = {
        'external_id': str(post['id']),
        'message': post['title'],
        'html_message': post['details'],
        'created_at': post['created_at'],
        'author': {'external_id': str(post['author_id']), 'name': 'community user'}
    }
    external_resources.append(resource)

Zendesk uses the resource objects in external_resources to create tickets. If no new posts were found, external_resources is an empty list.

The message value is used for the title of the new ticket. The html_message is used for the first comment in the ticket.

Next, the function sets the start time for the next pull request by finding the most recent created_at time in the list of external resources. It builds a list of datetime objects and gets the datetime with the greatest value (the most recent time). If no new posts were found, the current start_time is unchanged.

# get next start time
if external_resources:
    created_at_times = []
    for resource in external_resources:
        created_at_times.append(arrow.get(resource['created_at']))
    start_time = max(created_at_times)   # get most recent datetime in list

The start_time is assigned to the state parameter for storage and re-use in the next pull request:

# convert datetime to ISO 8601 string
state = {'start_time': start_time.isoformat()}

# convert state dict to JSON, then URL encode
state = urllib.parse.quote(json.dumps(state))

Finally, the function returns the data in a response format that Zendesk expects:

return {
    'external_resources': external_resources,
    'state': state
}

In turn, the Pull New Posts endpoint in service.py forwards the data in an HTTP response to Zendesk, which uses it to create tickets. See Channeling community posts to Support agents.

Replying to new posts

The Channelback Ticket Comment endpoint hands off the task of creating a post comment to a function called create_post_comment() in the community module:

external_id = community.create_post_comment(post_id, comment)

The function takes two arguments:

• post_id - a string containing the id of the originating post in the community
• comment - a string containing the agent's reply

To add the create_post_comment() function to the community module

Paste the following function into the community.py file and save the file.

...
def create_post_comment(post_id, comment):
    hc = get_hc_settings()
    url = f'{hc["base_url"]}/api/v2/community/posts/{post_id}/comments.json'
    data = {'comment': {'body': comment, 'author_id': hc['agent_id']}}
    auth = hc['credentials']
    headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}

    response = requests.post(url, json=data, auth=auth, headers=headers)
    if response.status_code != 201:
        error = f'{response.status_code}: {response.text}'
        print(f'Failed to create post comment with error {error}')
        return {'error': response.status_code}

    comment = response.json()['comment']
    return {'external_id': str(comment['id']), 'allow_channelback': False}

How it works

The function starts by retrieving your help center settings and preparing to make a request to the Create Comment endpoint in the Help Center API:

hc = get_hc_settings()
url = f'{hc["base_url"]}/api/v2/community/posts/{post_id}/comments.json'
data = {'comment': {'body': comment, 'author_id': hc['agent_id']}}
auth = hc['credentials']
headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}

The post_id and comment variables were passed to the function as the parent_id and message received from Zendesk.

Next, the function makes the request:

response = requests.post(url, json=data, auth=hc['credentials'], headers=headers)
    if response.status_code != 201:
        error = f'{response.status_code}: {response.text}'
        print(f'Failed to create post comment with error {error}')
        return {'error': response.status_code}

If the request doesn't return any errors, the function gets the comment id from the response, packages it in a response format that Zendesk expects, and returns it:

comment = response.json()['comment']
return {'external_id': str(comment['id']), 'allow_channelback': False}

In turn, the Channelback Ticket Comment endpoint in service.py forwards the data in an HTTP response to Zendesk. See Channeling agent comments to community posts.

In this article, you built a community module for your service that accomplishes two critical tasks:

• Gets community posts created after a specific start time
• Adds an agent comment in reply to a specific community post

In the next article, you'll create an admin interface in Zendesk Support for your channel.

Next part: Building the admin interface