Lab L2.2: DataMap Integration

🎯 Assignment: Accept this lab on GitHub Classroom
You’ll get your own repository with starter code, instructions, and automatic grading.


Duration	60 minutes
Prerequisites	Previous module completed

Objectives

Configure DataMap for external APIs
Handle API responses
Implement fallback behavior

How to Complete This Lab

Accept the Assignment — Click the GitHub Classroom link above
Clone Your Repo — git clone <your-repo-url>
Read the README — Your repo has detailed requirements and grading criteria
Write Your Code — Implement the solution in solution/agent.py
Test Locally — Use swaig-test to verify your agent works
Push to Submit — git push triggers auto-grading

Key Concepts

The following exercises walk through the concepts you’ll need. Your GitHub Classroom repo README has the specific requirements for grading.

Part 1: Basic DataMap (20 min)

Task

Create an agent that fetches user information from an API.

Starter Code

#!/usr/bin/env python3
"""Lab 2.2: DataMap integration."""

from signalwire_agents import AgentBase, DataMap, SwaigFunctionResult


class DataMapAgent(AgentBase):
    def __init__(self):
        super().__init__(name="datamap-agent")

        self.prompt_add_section(
            "Role",
            "You help look up user and post information from our database."
        )

        self.add_language("English", "en-US", "rime.spore")

        self._setup_datamaps()

    def _setup_datamaps(self):
        # TODO: Add DataMap for user lookup
        pass


if __name__ == "__main__":
    agent = DataMapAgent()
    agent.run()

Your Task

Add a DataMap that:

Fetches user by ID from https://jsonplaceholder.typicode.com/users/{user_id}
Returns the user’s name and email

Expected Result

def _setup_datamaps(self):
    # Create DataMap using builder pattern
    get_user = (
        DataMap("get_user")
        .description("Look up user information by ID")
        .parameter("user_id", "string", "The user ID (1-10)", required=True)
        .webhook("GET", "https://jsonplaceholder.typicode.com/users/${enc:args.user_id}")
        .output(SwaigFunctionResult("User ${response.name} has email ${response.email}")
            .update_global_data({
                "last_user_id": "${response.id}",
                "last_user_name": "${response.name}"
            })
        )
    )

    # Register the DataMap
    self.register_swaig_function(get_user.to_swaig_function())

Part 2: DataMap with POST (20 min)

Task

Create a DataMap that creates a new post (simulated).

Your Task

Add a DataMap that:

POSTs to https://jsonplaceholder.typicode.com/posts
Sends title and body as JSON
Confirms creation with the returned ID

Expected Result

# Create a POST DataMap
create_post = (
    DataMap("create_post")
    .description("Create a new blog post")
    .parameter("title", "string", "Post title", required=True)
    .parameter("body", "string", "Post content", required=True)
    .webhook(
        "POST",
        "https://jsonplaceholder.typicode.com/posts",
        headers={"Content-Type": "application/json"}
    )
    .body({
        "title": "${args.title}",
        "body": "${args.body}",
        "userId": 1
    })
    .output(SwaigFunctionResult("Post created with ID ${response.id}. Title: ${response.title}")
        .update_global_data({"last_post_id": "${response.id}"})
    )
)

self.register_swaig_function(create_post.to_swaig_function())

Part 3: Error Handling (20 min)

Task

Create a DataMap with proper error handling for invalid requests.

Your Task

Add a DataMap that:

Fetches a specific post by ID
Handles the case when post doesn’t exist (404)
Returns appropriate error messages

Expected Result

# Create a DataMap with error handling using fallback_output
get_post = (
    DataMap("get_post")
    .description("Get a blog post by ID")
    .parameter("post_id", "string", "The post ID", required=True)
    .webhook("GET", "https://jsonplaceholder.typicode.com/posts/${enc:args.post_id}")
    .output(SwaigFunctionResult("Post ${response.id}: ${response.title}")
        .update_global_data({"viewed_post_id": "${response.id}"})
    )
    .fallback_output(SwaigFunctionResult(
        "I couldn't find a post with that ID. Please try another."
    ))
)

self.register_swaig_function(get_post.to_swaig_function())

Testing

# Test the agent
swaig-test lab2_2_datamap.py --dump-swml

# List available tools
swaig-test lab2_2_datamap.py --list-tools

# Test get_user (user 1)
swaig-test lab2_2_datamap.py --exec get_user --user_id "1"

# Test create_post
swaig-test lab2_2_datamap.py --exec create_post \
  --title "Test Post" \
  --body "This is a test"

# Test get_post
swaig-test lab2_2_datamap.py --exec get_post --post_id "1"

# Test error handling (invalid ID)
swaig-test lab2_2_datamap.py --exec get_post --post_id "9999"

Validation Checklist

get_user fetches user data correctly
create_post sends POST request with JSON body
get_post handles both success and error cases
All DataMaps appear in SWML output
Variable substitution works correctly

Challenge Extension

Add a DataMap that:

Fetches all posts by a specific user
Returns a summary (count and first post title)
Use the /posts?userId={id} endpoint

Submission

Upload your completed lab2_2_datamap.py file.

Complete Agent Code

Click to reveal complete solution

#!/usr/bin/env python3
"""DataMap integration agent.

Lab 2.2 Deliverable: Demonstrates DataMap configuration for external
API calls with GET, POST, and error handling patterns.
"""

from signalwire_agents import AgentBase
from signalwire_agents.core.data_map import DataMap
from signalwire_agents.core.function_result import SwaigFunctionResult


class DataMapAgent(AgentBase):
    """Agent with DataMap integrations for external APIs."""

    def __init__(self):
        super().__init__(name="datamap-agent")

        self.prompt_add_section(
            "Role",
            "You help look up user and post information from our database."
        )

        self.prompt_add_section(
            "Capabilities",
            bullets=[
                "Look up user information by ID",
                "Create new blog posts",
                "Retrieve existing posts"
            ]
        )

        self.add_language("English", "en-US", "rime.spore")
        self._setup_datamaps()

    def _setup_datamaps(self):
        """Configure DataMaps for external API calls."""

        # GET request - User lookup
        get_user_dm = (
            DataMap("get_user")
            .description("Look up user information by ID")
            .parameter("user_id", "string", "The user ID (1-10)", required=True)
            .webhook(
                "GET",
                "https://jsonplaceholder.typicode.com/users/${args.user_id}"
            )
            .output(SwaigFunctionResult(
                "User ${name} has email ${email}"
            ))
            .fallback_output(SwaigFunctionResult(
                "Could not find user with that ID."
            ))
        )
        self.register_swaig_function(get_user_dm.to_swaig_function())

        # POST request - Create post
        create_post_dm = (
            DataMap("create_post")
            .description("Create a new blog post")
            .parameter("title", "string", "Post title", required=True)
            .parameter("body", "string", "Post content", required=True)
            .webhook(
                "POST",
                "https://jsonplaceholder.typicode.com/posts"
            )
            .body({
                "title": "${args.title}",
                "body": "${args.body}",
                "userId": 1
            })
            .output(SwaigFunctionResult(
                "Post created with ID ${id}. Title: ${title}"
            ))
            .fallback_output(SwaigFunctionResult(
                "Failed to create the post."
            ))
        )
        self.register_swaig_function(create_post_dm.to_swaig_function())

        # GET request with error handling
        get_post_dm = (
            DataMap("get_post")
            .description("Get a blog post by ID")
            .parameter("post_id", "string", "The post ID", required=True)
            .webhook(
                "GET",
                "https://jsonplaceholder.typicode.com/posts/${args.post_id}"
            )
            .output(SwaigFunctionResult(
                "Post ${id}: ${title}"
            ))
            .fallback_output(SwaigFunctionResult(
                "I couldn't find a post with that ID. Please try another."
            ))
        )
        self.register_swaig_function(get_post_dm.to_swaig_function())

        # Challenge: Get posts by user
        get_user_posts_dm = (
            DataMap("get_user_posts")
            .description("Get all posts by a specific user")
            .parameter("user_id", "string", "The user ID", required=True)
            .webhook(
                "GET",
                "https://jsonplaceholder.typicode.com/posts?userId=${args.user_id}"
            )
            .output(SwaigFunctionResult(
                "Found posts for user ${args.user_id}."
            ))
            .fallback_output(SwaigFunctionResult(
                "Could not retrieve posts for that user."
            ))
        )
        self.register_swaig_function(get_user_posts_dm.to_swaig_function())


if __name__ == "__main__":
    agent = DataMapAgent()
    agent.run()