API Reference

Complete reference for using AWS Knowledge Query in your Python code.

🎯 Overview

You can integrate AWS Knowledge Query functionality into your own Python applications using the provided classes and functions.

📦 Installation

pip install mcp httpx

🚀 Quick Start

Simple Query Function

import asyncio
from cli.ask_aws import ask_aws

# Ask a question
asyncio.run(ask_aws("How to create Lambda function", "general", 5))

Using the Assistant Class

import asyncio
from cli.interactive_aws import AWSAssistant

async def main():
    assistant = AWSAssistant()
    await assistant.connect()
    
    results = await assistant.ask("Lambda best practices", "general", 5)
    
    for result in results:
        print(f"Title: {result['title']}")
        print(f"URL: {result['url']}")
    
    await assistant.disconnect()

asyncio.run(main())

📚 API Reference

AWSAssistant Class

The main class for interacting with the AWS Knowledge MCP Server.

Constructor

from cli.interactive_aws import AWSAssistant

assistant = AWSAssistant()

Parameters: None

Attributes:

mcp_url (str): The MCP server endpoint
session (ClientSession): The MCP client session

Methods

connect()

Connect to the AWS Knowledge MCP Server.

await assistant.connect()

Returns: None

Raises: Exception if connection fails

Example:

assistant = AWSAssistant()
await assistant.connect()

disconnect()

Disconnect from the server and clean up resources.

await assistant.disconnect()

Returns: None

Example:

await assistant.disconnect()

ask(question, topic="general", limit=3)

Search AWS documentation.

results = await assistant.ask(
    question="Lambda best practices",
    topic="general",
    limit=5
)

Parameters:

question (str, required): Your question about AWS
topic (str, optional): Topic category (default: "general")
limit (int, optional): Number of results (default: 3, max: 10)

Returns: List of dictionaries with keys:

title (str): Document title
url (str): Documentation URL
context (str): Context snippet

Example:

results = await assistant.ask("boto3 S3 upload", "reference_documentation", 5)

for result in results:
    print(f"Title: {result['title']}")
    print(f"URL: {result['url']}")
    print(f"Context: {result['context'][:200]}...")

list_regions()

Get all AWS regions.

regions = await assistant.list_regions()

Returns: List of dictionaries with keys:

region_id (str): Region identifier (e.g., "us-east-1")
region_long_name (str): Full region name (e.g., "US East (N. Virginia)")

Example:

regions = await assistant.list_regions()

for region in regions:
    print(f"{region['region_id']}: {region['region_long_name']}")

check_availability(service, resource_type="product")

Check service availability across regions.

availability = await assistant.check_availability(
    service="AWS Lambda",
    resource_type="product"
)

Parameters:

service (str, required): Service name
resource_type (str, optional): Resource type (default: "product")

Returns: List of availability data

Example:

availability = await assistant.check_availability("AWS Lambda")
print(f"Found {len(availability)} availability records")

ask_aws Function

Simple function for one-off queries.

from cli.ask_aws import ask_aws

await ask_aws(question, topic="general", limit=5)

Parameters:

question (str, required): Your question
topic (str, optional): Topic category (default: "general")
limit (int, optional): Number of results (default: 5)

Returns: None (prints results to stdout)

Example:

import asyncio
from cli.ask_aws import ask_aws

asyncio.run(ask_aws("Lambda timeout error", "troubleshooting", 3))

🎯 Complete Examples

Example 1: Simple Search

import asyncio
from cli.interactive_aws import AWSAssistant

async def search_aws_docs():
    assistant = AWSAssistant()
    
    try:
        await assistant.connect()
        
        results = await assistant.ask("Lambda best practices", "general", 5)
        
        print(f"Found {len(results)} results:\n")
        for i, result in enumerate(results, 1):
            print(f"{i}. {result['title']}")
            print(f"   URL: {result['url']}")
            print(f"   Context: {result.get('context', 'N/A')[:150]}...\n")
    
    finally:
        await assistant.disconnect()

asyncio.run(search_aws_docs())

Example 2: Multiple Queries

import asyncio
from cli.interactive_aws import AWSAssistant

async def multiple_searches():
    assistant = AWSAssistant()
    await assistant.connect()
    
    try:
        # Search different topics
        queries = [
            ("Lambda best practices", "general"),
            ("boto3 S3 upload", "reference_documentation"),
            ("Lambda timeout error", "troubleshooting"),
        ]
        
        for question, topic in queries:
            print(f"\n{'='*60}")
            print(f"Query: {question} (Topic: {topic})")
            print('='*60)
            
            results = await assistant.ask(question, topic, 3)
            
            for result in results:
                print(f"- {result['title']}")
                print(f"  {result['url']}\n")
    
    finally:
        await assistant.disconnect()

asyncio.run(multiple_searches())

Example 3: Region Information

import asyncio
from cli.interactive_aws import AWSAssistant

async def get_regions():
    assistant = AWSAssistant()
    await assistant.connect()
    
    try:
        regions = await assistant.list_regions()
        
        print(f"Total AWS Regions: {len(regions)}\n")
        
        # Group by continent
        us_regions = [r for r in regions if r['region_id'].startswith('us-')]
        eu_regions = [r for r in regions if r['region_id'].startswith('eu-')]
        ap_regions = [r for r in regions if r['region_id'].startswith('ap-')]
        
        print("US Regions:")
        for region in us_regions:
            print(f"  {region['region_id']}: {region['region_long_name']}")
        
        print("\nEU Regions:")
        for region in eu_regions:
            print(f"  {region['region_id']}: {region['region_long_name']}")
        
        print("\nAP Regions:")
        for region in ap_regions:
            print(f"  {region['region_id']}: {region['region_long_name']}")
    
    finally:
        await assistant.disconnect()

asyncio.run(get_regions())

Example 4: Error Handling

import asyncio
from cli.interactive_aws import AWSAssistant

async def search_with_error_handling():
    assistant = AWSAssistant()
    
    try:
        await assistant.connect()
        print("✅ Connected to AWS Knowledge MCP Server")
        
        results = await assistant.ask("Lambda best practices", "general", 5)
        
        if not results:
            print("⚠️ No results found")
        else:
            print(f"✅ Found {len(results)} results")
            for result in results:
                print(f"- {result['title']}")
    
    except ConnectionError as e:
        print(f"❌ Connection error: {e}")
    except TimeoutError as e:
        print(f"❌ Timeout error: {e}")
    except Exception as e:
        print(f"❌ Unexpected error: {e}")
    
    finally:
        try:
            await assistant.disconnect()
            print("✅ Disconnected")
        except:
            pass

asyncio.run(search_with_error_handling())

Example 5: Custom Application

import asyncio
from cli.interactive_aws import AWSAssistant

class AWSDocsHelper:
    def __init__(self):
        self.assistant = AWSAssistant()
        self.connected = False
    
    async def __aenter__(self):
        await self.assistant.connect()
        self.connected = True
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.connected:
            await self.assistant.disconnect()
    
    async def search(self, question, topic="general", limit=5):
        """Search AWS documentation"""
        return await self.assistant.ask(question, topic, limit)
    
    async def get_regions(self):
        """Get all AWS regions"""
        return await self.assistant.list_regions()
    
    async def troubleshoot(self, error_message, limit=3):
        """Search troubleshooting docs"""
        return await self.assistant.ask(error_message, "troubleshooting", limit)
    
    async def get_api_docs(self, query, limit=5):
        """Search API documentation"""
        return await self.assistant.ask(query, "reference_documentation", limit)
    
    async def get_cdk_examples(self, query, limit=5):
        """Search CDK examples"""
        return await self.assistant.ask(query, "cdk_constructs", limit)

# Usage
async def main():
    async with AWSDocsHelper() as helper:
        # Search general docs
        results = await helper.search("Lambda best practices")
        print(f"General: {len(results)} results")
        
        # Troubleshoot an error
        results = await helper.troubleshoot("Lambda timeout error")
        print(f"Troubleshooting: {len(results)} results")
        
        # Get API docs
        results = await helper.get_api_docs("boto3 S3 upload")
        print(f"API docs: {len(results)} results")
        
        # Get CDK examples
        results = await helper.get_cdk_examples("Lambda CDK Python")
        print(f"CDK examples: {len(results)} results")
        
        # Get regions
        regions = await helper.get_regions()
        print(f"Regions: {len(regions)} total")

asyncio.run(main())

Example 6: Flask Integration

from flask import Flask, request, jsonify
import asyncio
from cli.interactive_aws import AWSAssistant

app = Flask(__name__)

@app.route('/api/search', methods=['POST'])
def search():
    data = request.json
    question = data.get('question')
    topic = data.get('topic', 'general')
    limit = data.get('limit', 5)
    
    if not question:
        return jsonify({"error": "Question required"}), 400
    
    # Run async function in sync context
    results = asyncio.run(search_aws(question, topic, limit))
    return jsonify({"results": results})

async def search_aws(question, topic, limit):
    assistant = AWSAssistant()
    await assistant.connect()
    try:
        return await assistant.ask(question, topic, limit)
    finally:
        await assistant.disconnect()

if __name__ == '__main__':
    app.run(port=5000)

📖 Topics Reference

Valid topic values:

Topic	Description
`general`	Best practices, architecture, tutorials
`reference_documentation`	API/SDK/CLI documentation
`troubleshooting`	Error messages and debugging
`cdk_docs`	CDK concepts and API reference
`cdk_constructs`	CDK code examples
`cloudformation`	CloudFormation templates
`current_awareness`	New features and announcements
`amplify_docs`	Amplify framework documentation

🔧 Configuration

MCP Server Endpoint

The default endpoint is:

mcp_url = "https://knowledge-mcp.global.api.aws"

To use a different endpoint:

assistant = AWSAssistant()
assistant.mcp_url = "https://your-custom-endpoint.com"
await assistant.connect()

Timeout Configuration

The default timeout is 60 seconds. To change it, modify the streamablehttp_client call in the source code.

🐛 Error Handling

Common Exceptions

try:
    await assistant.connect()
    results = await assistant.ask("Lambda", "general", 5)
except ConnectionError:
    print("Cannot connect to MCP server")
except TimeoutError:
    print("Request timed out")
except json.JSONDecodeError:
    print("Invalid response from server")
except Exception as e:
    print(f"Unexpected error: {e}")

Best Practices

Always use try/finally to ensure disconnection:

try:
    await assistant.connect()
    # Your code
finally:
    await assistant.disconnect()

Use context managers for automatic cleanup:

async with AWSDocsHelper() as helper:
    results = await helper.search("Lambda")

Handle empty results:

results = await assistant.ask("query", "general", 5)
if not results:
    print("No results found")

🔗 Related Guides

CLI Guide - Command-line usage
Interactive Guide - Interactive mode
Web UI Guide - Web interface
Troubleshooting - Common issues

Happy coding! 🚀

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

API Reference

🎯 Overview

📦 Installation

🚀 Quick Start

Simple Query Function

Using the Assistant Class

📚 API Reference

AWSAssistant Class

Constructor

Methods

connect()

disconnect()

ask(question, topic="general", limit=3)

list_regions()

check_availability(service, resource_type="product")

ask_aws Function

🎯 Complete Examples

Example 1: Simple Search

Example 2: Multiple Queries

Example 3: Region Information

Example 4: Error Handling

Example 5: Custom Application

Example 6: Flask Integration

📖 Topics Reference

🔧 Configuration

MCP Server Endpoint

Timeout Configuration

🐛 Error Handling

Common Exceptions

Best Practices

🔗 Related Guides

FilesExpand file tree

API_REFERENCE.md

Latest commit

History

API_REFERENCE.md

File metadata and controls

API Reference

🎯 Overview

📦 Installation

🚀 Quick Start

Simple Query Function

Using the Assistant Class

📚 API Reference

AWSAssistant Class

Constructor

Methods

connect()

disconnect()

ask(question, topic="general", limit=3)

list_regions()

check_availability(service, resource_type="product")

ask_aws Function

🎯 Complete Examples

Example 1: Simple Search

Example 2: Multiple Queries

Example 3: Region Information

Example 4: Error Handling

Example 5: Custom Application

Example 6: Flask Integration

📖 Topics Reference

🔧 Configuration

MCP Server Endpoint

Timeout Configuration

🐛 Error Handling

Common Exceptions

Best Practices

🔗 Related Guides