Implementation:Cohere ai Cohere python AwsChatModels

Knowledge Sources	Cohere Python SDK
Domains	SDK, API_Client
Last Updated	2026-02-15 14:00 GMT

Overview

AwsChatModels implements data models and streaming response handlers for chat interactions with AWS-deployed Cohere models via SageMaker and Bedrock.

Description

This module provides a complete set of classes for representing chat requests, responses, tool definitions, and streaming events when interacting with Cohere models deployed on AWS infrastructure (SageMaker and Bedrock). All classes inherit from CohereObject and most also extend dict for direct serialization compatibility.

Tool-related classes:

• ToolParameterDefinitionsValue -- Represents a single parameter definition for a tool, with type, description, and optional required flag.
• Tool -- Represents a tool that can be invoked by the model, with name, description, and parameter definitions.
• ToolCall -- Represents a tool invocation returned by the model, containing the tool name, parameters, and generation ID. Includes from_dict and from_list factory methods.

Chat response classes:

• Chat -- Represents a complete (non-streaming) chat response with response_id, generation_id, text, chat_history, token counts, citations, documents, search results, and tool calls. Includes a from_dict factory method.

Streaming classes:

• StreamEvent -- An enum defining the streaming event types: STREAM_START, SEARCH_QUERIES_GENERATION, SEARCH_RESULTS, TEXT_GENERATION, TOOL_CALLS_GENERATION, CITATION_GENERATION, and STREAM_END.
• StreamResponse -- Base class for all streaming event responses, carrying is_finished, event_type, and index.
• StreamStart, StreamTextGeneration, StreamCitationGeneration, StreamQueryGeneration, StreamSearchResults, StreamEnd, ChatToolCallsGenerationEvent -- Specialized streaming event classes for each event type.
• StreamingChat -- An iterable that processes raw AWS streaming payloads (from SageMaker or Bedrock), parsing JSON chunks and yielding typed StreamResponse objects. Accumulates final response state (text, citations, tool_calls, etc.) as the stream progresses.

Usage

Use these classes when building chat interactions with Cohere models deployed on AWS via SageMaker or Bedrock. The StreamingChat class is used to iterate over streaming responses, while Chat handles non-streaming responses.

Code Reference

Source Location

• Repository: Cohere Python SDK
• File: src/cohere/manually_maintained/cohere_aws/chat.py

Signature

class ToolParameterDefinitionsValue(CohereObject, dict):
    def __init__(self, type: str, description: str, required: Optional[bool] = None, **kwargs) -> None: ...

class Tool(CohereObject, dict):
    def __init__(self, name: str, description: str,
                 parameter_definitions: Optional[Dict[str, ToolParameterDefinitionsValue]] = None, **kwargs) -> None: ...

class ToolCall(CohereObject, dict):
    def __init__(self, name: str, parameters: Dict[str, Any], generation_id: str, **kwargs) -> None: ...
    @classmethod
    def from_dict(cls, tool_call_res: Dict[str, Any]) -> "ToolCall": ...
    @classmethod
    def from_list(cls, tool_calls_res: Optional[List[Dict[str, Any]]]) -> Optional[List["ToolCall"]]: ...

class Chat(CohereObject):
    def __init__(self, response_id: str, generation_id: str, text: str, ...) -> None: ...
    @classmethod
    def from_dict(cls, response: Dict[str, Any]) -> "Chat": ...

class StreamEvent(str, Enum):
    STREAM_START = "stream-start"
    SEARCH_QUERIES_GENERATION = "search-queries-generation"
    SEARCH_RESULTS = "search-results"
    TEXT_GENERATION = "text-generation"
    TOOL_CALLS_GENERATION = "tool-calls-generation"
    CITATION_GENERATION = "citation-generation"
    STREAM_END = "stream-end"

class StreamResponse(CohereObject):
    def __init__(self, is_finished: bool, event_type: Union[StreamEvent, str], index: Optional[int], **kwargs) -> None: ...

class StreamingChat(CohereObject):
    def __init__(self, stream_response, mode) -> None: ...
    def __iter__(self) -> Generator[StreamResponse, None, None]: ...

Import

from cohere.manually_maintained.cohere_aws.chat import (
    Tool,
    ToolCall,
    ToolParameterDefinitionsValue,
    Chat,
    StreamEvent,
    StreamResponse,
    StreamingChat,
)

I/O Contract

Inputs

Name	Type	Required	Description
name (Tool)	str	Yes	The name of the tool.
description (Tool)	str	Yes	A description of what the tool does.
parameter_definitions (Tool)	Optional[Dict[str, ToolParameterDefinitionsValue]]	No	Map of parameter names to their definitions.
name (ToolCall)	str	Yes	Name of the tool being called.
parameters (ToolCall)	Dict[str, Any]	Yes	Parameters to pass to the tool.
generation_id (ToolCall)	str	Yes	ID of the generation that produced this tool call.
response_id (Chat)	str	Yes	Unique identifier for the response.
generation_id (Chat)	str	Yes	Unique identifier for the generation.
text (Chat)	str	Yes	The generated text response.
stream_response (StreamingChat)	iterable	Yes	Raw AWS streaming response payload (from SageMaker or Bedrock).
mode (StreamingChat)	Mode	Yes	Deployment mode, either Mode.SAGEMAKER or Mode.BEDROCK, determining payload key structure.

Outputs

Name	Type	Description
Chat	Chat	A complete chat response object with text, tool_calls, citations, documents, search results, and metadata.
StreamingChat (iterable)	Generator[StreamResponse, None, None]	Yields typed streaming event objects (StreamStart, StreamTextGeneration, StreamEnd, etc.) as they arrive.
ToolCall.from_dict	ToolCall	A ToolCall instance constructed from a raw dictionary.
ToolCall.from_list	Optional[List[ToolCall]]	A list of ToolCall instances, or None if input is None or not a list.

Usage Examples

from cohere.manually_maintained.cohere_aws.chat import Chat, Tool, ToolParameterDefinitionsValue

# Define a tool
search_tool = Tool(
    name="web_search",
    description="Searches the web for relevant information",
    parameter_definitions={
        "query": ToolParameterDefinitionsValue(
            type="string",
            description="The search query",
            required=True,
        )
    },
)

# Parse a non-streaming response from a raw dict
raw_response = {
    "response_id": "abc-123",
    "generation_id": "gen-456",
    "text": "Here is the answer to your question.",
    "finish_reason": "COMPLETE",
    "token_count": {"prompt_tokens": 10, "response_tokens": 15, "total_tokens": 25},
}
chat_response = Chat.from_dict(raw_response)
print(chat_response.text)  # "Here is the answer to your question."

# Iterate over a streaming response (AWS Bedrock example)
from cohere.manually_maintained.cohere_aws.chat import StreamingChat
from cohere.manually_maintained.cohere_aws.mode import Mode

streaming_chat = StreamingChat(stream_response=bedrock_response_stream, mode=Mode.BEDROCK)
for event in streaming_chat:
    if event.event_type == "text-generation":
        print(event.text, end="")

Related Pages

• Environment:Cohere_ai_Cohere_python_Python_SDK_Runtime