get_connection
Retrieve Airflow connection details by ID to access database, API, and service configurations for workflow automation.
Instructions
Get a connection by ID
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| conn_id | Yes |
Input Schema (JSON Schema)
{
"properties": {
"conn_id": {
"title": "Conn Id",
"type": "string"
}
},
"required": [
"conn_id"
],
"type": "object"
}
Implementation Reference
- src/airflow/connection.py:72-74 (handler)The handler function that executes the get_connection tool logic by calling the Airflow ConnectionApi to retrieve a connection by ID and formatting the response as MCP TextContent.async def get_connection(conn_id: str) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]: response = connection_api.get_connection(connection_id=conn_id) return [types.TextContent(type="text", text=str(response.to_dict()))]
- src/airflow/connection.py:11-20 (registration)Function providing the registration tuple for get_connection (along with other connection tools), including name, description, and read-only status.def get_all_functions() -> list[tuple[Callable, str, str, bool]]: """Return list of (function, name, description, is_read_only) tuples for registration.""" return [ (list_connections, "list_connections", "List all connections", True), (create_connection, "create_connection", "Create a connection", False), (get_connection, "get_connection", "Get a connection by ID", True), (update_connection, "update_connection", "Update a connection by ID", False), (delete_connection, "delete_connection", "Delete a connection by ID", False), (test_connection, "test_connection", "Test a connection", True), ]
- src/main.py:78-92 (registration)Main registration loop that loads functions from connection module (via APITYPE_TO_FUNCTIONS) and registers them as MCP tools using app.add_tool, including get_connection.for api in apis: logging.debug(f"Adding API: {api}") get_function = APITYPE_TO_FUNCTIONS[APIType(api)] try: functions = get_function() except NotImplementedError: continue # Filter functions for read-only mode if requested if read_only: functions = filter_functions_for_read_only(functions) for func, name, description, *_ in functions: app.add_tool(func, name=name, description=description)